use crate::stores::database::DatabaseStore;
use bytes::{BufMut, Bytes, BytesMut};
use flexbuffers::Reader;
use ordinary_config::DatabaseModelConfig;
use ordinary_types::Kind;
use saferlmdb::{WriteTransaction, put};
use thiserror::Error;
use tracing::instrument;
#[derive(Error, Debug)]
pub enum DeleteError {
#[error("no model for idx {0}")]
ModelNotFound(u8),
#[error("primary key not present in database")]
ItemNotFound,
#[error("item cannot be deleted as children still hold back reference")]
ChildBackReferencesRemain,
#[error("value has already been indexed")]
IndexCollision,
#[error("invalid kind for index: {0}")]
InvalidIndexKind(Kind),
#[error("invalid kind for query: {0}")]
InvalidQueryKind(Kind),
#[error("item size exceeds limits")]
ExceedsSizeLimits,
#[error("(FlexBuffers) {0}")]
FlexBuffersReaderError(#[from] flexbuffers::ReaderError),
#[error("(LMDB) {0}")]
LmdbError(#[from] saferlmdb::Error),
#[error(transparent)]
Anyhow(#[from] anyhow::Error),
#[error(transparent)]
TryFromIntError(#[from] std::num::TryFromIntError),
}
#[allow(clippy::too_many_lines)]
#[instrument(skip_all, fields(i = model_config.idx, nm = tracing::field::display(&model_config.name), uuid = tracing::field::display(uuid::Uuid::from_bytes(uuid))))]
pub(crate) fn delete(
database: &DatabaseStore,
model_config: &DatabaseModelConfig,
uuid: [u8; 16],
txn: Option<WriteTransaction<'_>>,
) -> anyhow::Result<(), DeleteError> {
let res = inner(database, model_config, uuid, txn);
if let Err(err) = &res {
#[allow(clippy::single_match_else)]
match err {
DeleteError::ItemNotFound => tracing::warn!("item not found"),
_ => tracing::error!("{err}"),
}
}
res
}
#[inline]
#[allow(clippy::too_many_lines)]
fn inner(
database: &DatabaseStore,
model_config: &DatabaseModelConfig,
uuid: [u8; 16],
txn: Option<WriteTransaction>,
) -> Result<(), DeleteError> {
let mut index_key = BytesMut::new();
index_key.put_u8(model_config.idx);
index_key.put_u8(0);
index_key.put(uuid.as_slice());
let txn = match txn {
Some(txn) => txn,
None => WriteTransaction::new(database.env.clone())?,
};
{
let mut access = txn.access();
let item_key = match access.get::<[u8], [u8]>(&database.index_db, index_key.as_ref()) {
Ok(ik) => Bytes::copy_from_slice(ik),
Err(_) => return Err(DeleteError::ItemNotFound),
};
let item = access.get::<[u8], [u8]>(&database.item_db, item_key.as_ref())?;
let item = Bytes::copy_from_slice(item);
let root_vec = Reader::get_root(item.as_ref())?.as_vector();
for field in &model_config.fields {
let reader = root_vec.idx(field.idx as usize);
if let Kind::Ref {
model: _,
field: _,
many,
} = &field.kind
{
if let Some((_ref_model_idx, ref_field_idx, ref_model_fields, ref_is_many)) =
database.cross_refs.get(&(model_config.idx, field.idx))
{
if many == &Some(true) && *ref_is_many {
for ref_reader in &reader.as_vector() {
let ref_item_id_bytes = ref_reader.as_blob().0;
let ref_item =
access.get::<[u8], [u8]>(&database.item_db, ref_item_id_bytes)?;
let root_vec = Reader::get_root(ref_item)?.as_vector();
let mut builder =
flexbuffers::Builder::new(&flexbuffers::BuilderOptions::SHARE_NONE);
let mut builder_vec = builder.start_vector();
builder_vec.push(root_vec.idx(0).as_blob());
for ref_field in ref_model_fields {
if &ref_field.idx == ref_field_idx {
let mut curr_refs = builder_vec.start_vector();
for curr_ref in
&root_vec.idx(ref_field.idx as usize).as_vector()
{
let curr_ref = curr_ref.as_blob();
let curr_ref_bytes = curr_ref.0;
if curr_ref_bytes != item_key {
curr_refs.push(curr_ref);
}
}
curr_refs.end_vector();
} else {
ref_field.kind.copy_to(
&root_vec.idx(ref_field.idx as usize),
&mut builder_vec,
None,
)?;
}
}
builder_vec.end_vector();
let updated_ref_item = builder.view();
access.put(
&database.item_db,
ref_item_id_bytes,
updated_ref_item,
&put::Flags::empty(),
)?;
}
} else if *ref_is_many {
let ref_item_id_bytes = reader.as_blob().0;
let ref_item =
access.get::<[u8], [u8]>(&database.item_db, ref_item_id_bytes)?;
let root_vec = Reader::get_root(ref_item)?.as_vector();
let mut builder =
flexbuffers::Builder::new(&flexbuffers::BuilderOptions::SHARE_NONE);
let mut builder_vec = builder.start_vector();
builder_vec.push(root_vec.idx(0).as_blob());
for ref_field in ref_model_fields {
if &ref_field.idx == ref_field_idx {
let mut curr_refs = builder_vec.start_vector();
for curr_ref in &root_vec.idx(ref_field.idx as usize).as_vector() {
let curr_ref = curr_ref.as_blob();
let curr_ref_bytes = curr_ref.0;
if curr_ref_bytes != item_key {
curr_refs.push(curr_ref);
}
}
curr_refs.end_vector();
} else {
ref_field.kind.copy_to(
&root_vec.idx(ref_field.idx as usize),
&mut builder_vec,
None,
)?;
}
}
builder_vec.end_vector();
let updated_ref_item = builder.view();
access.put(
&database.item_db,
ref_item_id_bytes,
updated_ref_item,
&put::Flags::empty(),
)?;
} else if !reader.as_vector().is_empty() {
return Err(DeleteError::ChildBackReferencesRemain);
}
}
} else {
if field.indexed == Some(true) {
let mut key = BytesMut::new();
key.put_u8(model_config.idx);
key.put_u8(field.idx);
match field.kind {
Kind::String => {
key.put(reader.as_str().as_bytes());
}
Kind::Uuid => {
key.put(reader.as_blob().0);
}
_ => {}
}
access.del_key(&database.index_db, key.as_ref())?;
}
if field.queryable == Some(true) {
let mut key = BytesMut::new();
key.put_u8(model_config.idx);
key.put_u8(field.idx);
match field.kind {
Kind::String => key.put(reader.as_str().as_bytes()),
Kind::Uuid => {
key.put(reader.as_blob().0);
}
_ => return Err(DeleteError::InvalidQueryKind(field.kind.clone())),
}
let mut builder =
flexbuffers::Builder::new(&flexbuffers::BuilderOptions::SHARE_NONE);
let mut builder_vec = builder.start_vector();
if let Ok(list) = access.get::<[u8], [u8]>(&database.query_db, key.as_ref()) {
let root = Reader::get_root(list)?;
for val in &root.as_vector() {
let key = val.as_blob();
let key_bytes = key.0;
if key_bytes != item_key {
builder_vec.push(key);
}
}
}
builder_vec.end_vector();
let updated_list = builder.view();
access.put(
&database.query_db,
key.as_ref(),
updated_list,
&put::Flags::empty(),
)?;
}
}
}
access.del_key(&database.item_db, item_key.as_ref())?;
access.del_key(&database.index_db, index_key.as_ref())?;
}
txn.commit()?;
Ok(())
}