// Copyright (C) 2026 Ordinary Labs, LLC.
//
// SPDX-License-Identifier: AGPL-3.0-only

use crate::{
    Accessor, RefDepth, Transaction, add_field_to_doc_flexbuffer, build_schema_for_fields,
    field_to_bytes, get_term, push_field_from_bytes,
};
use anyhow::bail;
use bytes::{BufMut, Bytes, BytesMut};
use chacha20poly1305::{
    KeyInit, XChaCha20Poly1305, XNonce,
    aead::{Aead, Generate},
};
use flexbuffers::Reader;
use hashbrown::{HashMap, HashSet};
use ordinary_config::{ModelConfig, ModelLimits, UuidVersion};
use ordinary_types::{Field, Kind, TimeUnit};
use parking_lot::Mutex;
use saferlmdb::{
    self as lmdb, Database, DatabaseOptions, Environment, ReadTransaction, WriteTransaction, put,
};
use std::path::Path;
use std::sync::Arc;
use tantivy::collector::TopDocs;
use tantivy::directory::MmapDirectory;
use tantivy::query::FuzzyTermQuery;
use tantivy::schema::{Schema, Value as TantivyValue};
use tantivy::{Index, IndexWriter, TantivyDocument, Term};
use tracing::instrument;
use uuid::Uuid;

pub struct ModelStore {
    pub limits: ModelLimits,

    env: Arc<Environment>,

    /// Vec of auto incrementing ints per-model
    auto_ints: Vec<Arc<Mutex<usize>>>,

    /// key([`model_byte`][int_id]) -> value(flexbuffer encoded)
    item_db: Arc<Database<'static>>,
    /// key([`model_idx`][field_idx][`indexed_value`]) -> value([kind][int_id])
    index_db: Arc<Database<'static>>,
    /// key([`model_idx`][field_idx][`indexed_value`]) -> value(Vec<[kind][int_id]>)
    query_db: Arc<Database<'static>>,

    /// used for server-side field encryption
    encryption_key: [u8; 32],

    model_configs: Vec<ModelConfig>,
    model_config_map: HashMap<String, ModelConfig>, // todo: switch to `rustc_hash`

    /// only for refs that are many. only a single many reference
    /// to another model can exist on a given model.
    ///
    /// (model, field) -> (model idx, field idx, fields, `is_many`)
    #[allow(clippy::type_complexity)]
    cross_refs: HashMap<(u8, u8), (u8, u8, Vec<Field>, bool)>, // todo: switch to `rustc_hash`

    model_search_indexes: Vec<Option<(Schema, Index, tantivy::IndexReader)>>,

    log_size: bool,
}

impl ModelStore {
    #[allow(clippy::too_many_lines)]
    pub fn new(
        limits: ModelLimits,
        mut model_configs: Vec<ModelConfig>,
        encryption_key: [u8; 32],
        env: &Arc<Environment>,
        search_dir: impl AsRef<Path>,
        log_size: bool,
    ) -> anyhow::Result<Self> {
        let item_db = Arc::new(Database::open(
            env.clone(),
            Some("item"),
            &DatabaseOptions::new(lmdb::db::Flags::CREATE),
        )?);
        let index_db = Arc::new(Database::open(
            env.clone(),
            Some("index"),
            &DatabaseOptions::new(lmdb::db::Flags::CREATE),
        )?);
        let query_db = Arc::new(Database::open(
            env.clone(),
            Some("query"),
            &DatabaseOptions::new(lmdb::db::Flags::CREATE),
        )?);

        let mut auto_ints = vec![];
        let mut cross_refs = HashMap::new();

        model_configs.sort_by_key(|a| a.idx);
        for config in &mut model_configs {
            config.fields.sort_by_key(|a| a.idx);
        }

        let mut model_config_map = HashMap::new();
        let mut model_search_indexes = vec![];

        for config in &model_configs {
            model_config_map.insert(config.name.clone(), config.clone());

            if limits.search_enabled {
                let mut schema_builder = Schema::builder();

                let should_build =
                    build_schema_for_fields(&mut schema_builder, &config.fields, true);

                if should_build {
                    let schema = schema_builder.build();

                    let index_dir = search_dir
                        .as_ref()
                        .join("model")
                        .join(config.idx.to_string());
                    std::fs::create_dir_all(&index_dir)?;

                    let dir = MmapDirectory::open(index_dir)?;
                    let index = Index::open_or_create(dir, schema.clone())?;

                    let reader = index
                        .reader_builder()
                        .reload_policy(tantivy::ReloadPolicy::OnCommitWithDelay);

                    model_search_indexes.push(Some((schema, index, reader.try_into()?)));
                } else {
                    model_search_indexes.push(None);
                    drop(schema_builder);
                }
            }

            let txn = ReadTransaction::new(env.clone())?;
            let access = txn.access();

            let mut item_cursor = txn.cursor(item_db.clone())?;

            match item_cursor.seek_range_k::<[u8], [u8]>(&access, &[config.idx]) {
                Ok((key, _)) => {
                    let model_idx = key[0];

                    if model_idx == config.idx {
                        let mut curr = &key[1..];

                        loop {
                            if let Ok((key, _)) = item_cursor.next::<[u8], [u8]>(&access) {
                                if key[0] == config.idx {
                                    curr = &key[1..];
                                } else {
                                    match curr.len() {
                                        1 => auto_ints.push(Arc::new(Mutex::new(curr[0] as usize))),
                                        2 => auto_ints.push(Arc::new(Mutex::new(
                                            u16::from_be_bytes(curr.try_into()?) as usize,
                                        ))),
                                        4 => auto_ints.push(Arc::new(Mutex::new(
                                            u32::from_be_bytes(curr.try_into()?) as usize,
                                        ))),
                                        _ => auto_ints.push(Arc::new(Mutex::new(usize::try_from(
                                            u64::from_be_bytes(curr.try_into()?),
                                        )?))),
                                    }
                                    break;
                                }
                            } else {
                                match curr.len() {
                                    1 => auto_ints.push(Arc::new(Mutex::new(curr[0] as usize))),
                                    2 => auto_ints.push(Arc::new(Mutex::new(u16::from_be_bytes(
                                        curr.try_into()?,
                                    )
                                        as usize))),
                                    4 => auto_ints.push(Arc::new(Mutex::new(u32::from_be_bytes(
                                        curr.try_into()?,
                                    )
                                        as usize))),
                                    _ => auto_ints.push(Arc::new(Mutex::new(usize::try_from(
                                        u64::from_be_bytes(curr.try_into()?),
                                    )?))),
                                }
                                break;
                            }
                        }
                    } else {
                        auto_ints.push(Arc::new(Mutex::new(0)));
                    }
                }
                Err(_) => {
                    auto_ints.push(Arc::new(Mutex::new(0)));
                }
            }

            for field in &config.fields {
                if let Kind::Ref {
                    model,
                    field: _ref_field,
                    many: _many,
                } = &field.kind
                {
                    for ref_config in &model_configs {
                        if ref_config.name == *model {
                            for ref_field in &ref_config.fields {
                                if let Kind::Ref {
                                    model: back_ref_model,
                                    many: back_ref_many,
                                    field: _back_ref_field,
                                } = &ref_field.kind
                                    && *back_ref_model == config.name
                                {
                                    cross_refs.insert(
                                        (config.idx, field.idx),
                                        (
                                            ref_config.idx,
                                            ref_field.idx,
                                            ref_config.fields.clone(),
                                            back_ref_many == &Some(true),
                                        ),
                                    );
                                }
                            }
                        }
                    }
                }
            }
        }

        Ok(Self {
            limits,
            env: env.clone(),

            auto_ints,
            encryption_key,

            item_db,
            index_db,
            query_db,

            model_configs,
            model_config_map,

            cross_refs,

            model_search_indexes,

            log_size,
        })
    }

    fn encrypt_compress_copy_to(
        &self,
        field: &Field,
        reader: &Reader<&[u8]>,
        item: &mut flexbuffers::VectorBuilder,
    ) -> anyhow::Result<()> {
        if field.encrypted == Some(true) {
            if field.indexed == Some(true)
                || field.queryable == Some(true)
                || field.searchable == Some(true)
            {
                bail!("cannot encrypt indexed/queryable/searchable fields");
            }

            let cipher = XChaCha20Poly1305::new(&self.encryption_key.into());
            let nonce = XNonce::generate();

            let plaintext = if field.compressed == Some(true) {
                let bytes = field_to_bytes(field, reader);
                Bytes::copy_from_slice(
                    zstd::stream::encode_all(std::io::Cursor::new(bytes.as_ref()), 4)?.as_slice(),
                )
            } else {
                field_to_bytes(field, reader)
            };

            match cipher.encrypt(&nonce, plaintext.as_ref()) {
                Ok(mut v) => {
                    if v.len() as u64 > self.limits.max_field_size {
                        if field.compressed == Some(true) {
                            bail!(
                                "i = {}, nm = {}, encrypted + compressed field exceeds size limits",
                                field.idx,
                                field.name
                            )
                        }

                        bail!(
                            "i = {}, nm = {}, encrypted field exceeds size limits",
                            field.idx,
                            field.name
                        )
                    }

                    v.extend_from_slice(&nonce);
                    item.push(flexbuffers::Blob(v.as_ref()));
                }
                Err(err) => {
                    tracing::error!(%err);
                }
            }
        } else if field.compressed == Some(true) {
            let bytes = field_to_bytes(field, reader);
            let compressed = zstd::stream::encode_all(std::io::Cursor::new(bytes.as_ref()), 4)?;

            if compressed.len() as u64 > self.limits.max_field_size {
                bail!(
                    "i = {}, nm = {}, compressed field exceeds size limits",
                    field.idx,
                    field.name
                );
            }

            item.push(flexbuffers::Blob(compressed.as_slice()));
        } else {
            let mut size = 0;
            field.kind.copy_to(reader, item, Some(&mut size))?;

            if size as u64 > self.limits.max_field_size {
                bail!(
                    "i = {}, nm = {}, field exceeds size limits",
                    field.idx,
                    field.name
                );
            }
        }

        Ok(())
    }

    #[allow(clippy::too_many_lines)]
    #[instrument(skip_all, fields(i, nm, uuid), err)]
    pub fn insert_model(
        &self,
        model_idx: u8,
        payload: &[u8],
        txn: Option<WriteTransaction<'_>>,
    ) -> anyhow::Result<[u8; 16]> {
        if let Some(model) = self.model_configs.get(model_idx as usize) {
            let curr_span = tracing::Span::current();
            curr_span.record("i", model.idx);
            curr_span.record("nm", tracing::field::display(&model.name));

            let mut size: isize = 0;
            let mut inserts = 0;

            let root_vec = Reader::get_root(payload)?.as_vector();

            let mut builder = flexbuffers::Builder::new(&flexbuffers::BuilderOptions::SHARE_NONE);
            let mut item = builder.start_vector();

            let uuid = if let Some(uuid_version) = &model.uuid {
                match uuid_version {
                    UuidVersion::V4 => {
                        let uuid = Uuid::new_v4();
                        curr_span.record("uuid", uuid.to_string());

                        *uuid.as_bytes()
                    }
                    UuidVersion::V7 => {
                        let uuid = Uuid::now_v7();
                        curr_span.record("uuid", uuid.to_string());

                        *uuid.as_bytes()
                    }
                }
            } else {
                let uuid = Uuid::new_v4();
                curr_span.record("uuid", uuid.to_string());

                *uuid.as_bytes()
            };

            let mut model_prefixed_uuid = BytesMut::new();

            model_prefixed_uuid.put_u8(model_idx);
            // 0 field
            model_prefixed_uuid.put_u8(0);
            model_prefixed_uuid.put(uuid.as_ref());

            let curr = if let Some(auto_int) = self.auto_ints.get(model_idx as usize) {
                let mut auto_int = auto_int.lock();
                *auto_int += 1;

                *auto_int
            } else {
                bail!("no auto int");
            };

            let mut item_key = BytesMut::new();

            item_key.put_u8(model_idx);

            match curr {
                0..=255 => {
                    item_key.put_u8(u8::try_from(curr)?);
                }
                256..=65_535 => {
                    item_key.put(u16::try_from(curr)?.to_be_bytes().as_slice());
                }
                65_536..=4_294_967_295 => {
                    item_key.put(u32::try_from(curr)?.to_be_bytes().as_slice());
                }
                _ => {
                    item_key.put((curr as u64).to_be_bytes().as_slice());
                }
            }

            item.push(flexbuffers::Blob(uuid.as_slice()));

            let mut search = None;

            if self.limits.search_enabled
                && let Some(searchable) = self.model_search_indexes.get(model_idx as usize)
                && let Some((schema, index, _reader)) = searchable
            {
                let mut doc = TantivyDocument::default();
                let id_field = schema.get_field("0")?;
                doc.add_bytes(id_field, item_key.as_ref());

                search = Some((doc, schema, index));
            }

            let txn = match txn {
                Some(txn) => txn,
                None => WriteTransaction::new(self.env.clone())?,
            };

            {
                let mut access = txn.access();

                access.put(
                    &self.index_db,
                    model_prefixed_uuid.as_ref(),
                    item_key.as_ref(),
                    &put::Flags::empty(),
                )?;
                inserts += 1;
                size +=
                    isize::try_from(model_prefixed_uuid.len())? + isize::try_from(item_key.len())?;

                for field in &model.fields {
                    let reader = root_vec.idx(field.idx as usize - 1);

                    if let Kind::Ref {
                        model: _,
                        field: _,
                        many,
                    } = &field.kind
                    {
                        if let Some((ref_model_idx, ref_field_idx, ref_model_fields, ref_is_many)) =
                            self.cross_refs.get(&(model_idx, field.idx))
                        {
                            // many-to-many
                            if many == &Some(true) && *ref_is_many {
                                let mut refs = item.start_vector();
                                let mut seen_uuids = HashSet::new();

                                for uuid in &reader.as_vector() {
                                    let uuid = uuid.as_blob();
                                    let uuid_bytes = uuid.0;
                                    let mut lookup = BytesMut::new();

                                    lookup.put_u8(*ref_model_idx);
                                    lookup.put_u8(0);

                                    lookup.put(uuid_bytes);

                                    if seen_uuids.contains(uuid_bytes) {
                                        continue;
                                    }
                                    seen_uuids.insert(Bytes::copy_from_slice(uuid_bytes));

                                    let ref_item_id = access
                                        .get::<[u8], [u8]>(&self.index_db, lookup.as_ref())?;
                                    let ref_item_id = Bytes::copy_from_slice(ref_item_id);

                                    refs.push(flexbuffers::Blob(ref_item_id.as_ref()));

                                    // update foreign
                                    let ref_item = access
                                        .get::<[u8], [u8]>(&self.item_db, ref_item_id.as_ref())?;

                                    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(uuid);

                                    for ref_field in ref_model_fields {
                                        if &ref_field.idx == ref_field_idx {
                                            let mut curr_refs = builder_vec.start_vector();

                                            // pushes self ref to front of foreign's refs
                                            curr_refs.push(flexbuffers::Blob(item_key.as_ref()));

                                            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();
                                    let ref_item_len = isize::try_from(ref_item.len())?;

                                    access.put(
                                        &self.item_db,
                                        ref_item_id.as_ref(),
                                        updated_ref_item,
                                        &put::Flags::empty(),
                                    )?;

                                    inserts += 1;
                                    size += isize::try_from(updated_ref_item.len())? - ref_item_len;
                                }

                                refs.end_vector();
                            // one-of-many (on another record)
                            } else if *ref_is_many {
                                let uuid = reader.as_blob();
                                let uuid_bytes = uuid.0;
                                let mut lookup = BytesMut::new();

                                lookup.put_u8(*ref_model_idx);
                                lookup.put_u8(0);

                                lookup.put(uuid_bytes);

                                let ref_item_id =
                                    access.get::<[u8], [u8]>(&self.index_db, lookup.as_ref())?;

                                let ref_item_id = Bytes::copy_from_slice(ref_item_id);
                                item.push(flexbuffers::Blob(ref_item_id.as_ref()));

                                // update foreign
                                let ref_item = access
                                    .get::<[u8], [u8]>(&self.item_db, ref_item_id.as_ref())?;

                                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(uuid);

                                for ref_field in ref_model_fields {
                                    if &ref_field.idx == ref_field_idx {
                                        let mut curr_refs = builder_vec.start_vector();

                                        // pushes self ref to front of foreign's refs
                                        curr_refs.push(flexbuffers::Blob(item_key.as_ref()));

                                        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();
                                let ref_item_len = isize::try_from(ref_item.len())?;

                                access.put(
                                    &self.item_db,
                                    ref_item_id.as_ref(),
                                    updated_ref_item,
                                    &put::Flags::empty(),
                                )?;

                                inserts += 1;
                                size += isize::try_from(updated_ref_item.len())? - ref_item_len;

                            // many-to-one
                            } else {
                                let empty = item.start_vector();
                                empty.end_vector();
                            }
                        }
                    } else {
                        if field.indexed == Some(true) {
                            let mut key = BytesMut::new();
                            key.put_u8(model_idx);
                            key.put_u8(field.idx);

                            match field.kind {
                                Kind::String | Kind::Url => {
                                    key.put(reader.as_str().as_bytes());
                                }
                                Kind::Uuid => {
                                    key.put(reader.as_blob().0);
                                }
                                _ => {
                                    bail!("cannot index on {:?}", field.kind);
                                }
                            }

                            if access
                                .get::<[u8], [u8]>(&self.index_db, key.as_ref())
                                .is_ok()
                            {
                                bail!("value has already been indexed");
                            }

                            access.put(
                                &self.index_db,
                                key.as_ref(),
                                item_key.as_ref(),
                                &put::Flags::empty(),
                            )?;
                            inserts += 1;
                            size += isize::try_from(key.len())? + isize::try_from(item_key.len())?;
                        }

                        if field.queryable == Some(true) {
                            let mut key = BytesMut::new();
                            key.put_u8(model_idx);
                            key.put_u8(field.idx);

                            match &field.kind {
                                Kind::Bool => {
                                    if reader.as_bool() {
                                        key.put_u8(1);
                                    } else {
                                        key.put_u8(0);
                                    }
                                }
                                Kind::String | Kind::Url => key.put(reader.as_str().as_bytes()),
                                Kind::Uuid => key.put(reader.as_blob().0),
                                Kind::Timestamp { unit } => match &unit {
                                    TimeUnit::Seconds => key.put_i64(reader.as_i64()),
                                },
                                Kind::U8 => key.put_u8(reader.as_u8()),
                                Kind::U16 => key.put_u16(reader.as_u16()),
                                Kind::U32 => key.put_u32(reader.as_u32()),
                                Kind::U64 => key.put_u64(reader.as_u64()),
                                Kind::I8 => key.put_i8(reader.as_i8()),
                                Kind::I16 => key.put_i16(reader.as_i16()),
                                Kind::I32 => key.put_i32(reader.as_i32()),
                                Kind::I64 => key.put_i64(reader.as_i64()),
                                Kind::F32 => key.put_f32(reader.as_f32()),
                                Kind::F64 => key.put_f64(reader.as_f64()),
                                _ => {
                                    bail!("cannot query on {:?}", field.kind);
                                }
                            }

                            let mut builder =
                                flexbuffers::Builder::new(&flexbuffers::BuilderOptions::SHARE_NONE);
                            let mut builder_vec = builder.start_vector();

                            builder_vec.push(flexbuffers::Blob(item_key.as_ref()));

                            let list_len = if let Ok(list) =
                                access.get::<[u8], [u8]>(&self.query_db, key.as_ref())
                            {
                                let root = Reader::get_root(list)?;
                                for val in &root.as_vector() {
                                    builder_vec.push(val.as_blob());
                                }

                                list.len()
                            } else {
                                0
                            };

                            builder_vec.end_vector();

                            let updated_list = builder.view();

                            access.put(
                                &self.query_db,
                                key.as_ref(),
                                updated_list,
                                &put::Flags::empty(),
                            )?;
                            inserts += 1;
                            size +=
                                isize::try_from(updated_list.len())? - isize::try_from(list_len)?;
                        }

                        if field.searchable == Some(true)
                            && let Some((doc, schema, _index)) = &mut search
                        {
                            let tantivy_field = schema.get_field(&field.idx.to_string())?;
                            add_field_to_doc_flexbuffer(doc, field, tantivy_field, &reader);
                        }

                        self.encrypt_compress_copy_to(field, &reader, &mut item)?;
                    }
                }

                item.end_vector();

                let item = builder.view();

                if item.len() as u64 > self.limits.max_item_size {
                    bail!("item size exceeds limits");
                }

                access.put(&self.item_db, item_key.as_ref(), item, &put::Flags::empty())?;

                inserts += 1;
                size += isize::try_from(item_key.len() + item.len())?;
            }

            txn.commit()?;

            if self.log_size {
                if size > 0 {
                    tracing::info!(inserts, size = %bytesize::ByteSize(size as u64).display().si_short());
                } else {
                    tracing::info!(
                        inserts,
                        size =
                            format!("-{}", bytesize::ByteSize(-size as u64).display().si_short())
                    );
                }
            } else {
                tracing::info!(inserts);
            }

            if let Some((doc, _schema, index)) = search {
                // todo: see if this is reusable so it doesn't have to be allocated on every write
                let mut index_writer: IndexWriter = index.writer(15_000_000)?;
                index_writer.add_document(doc)?;

                index_writer.commit()?;
                index_writer.wait_merging_threads()?;
            }

            Ok(uuid)
        } else {
            bail!("no model at that index")
        }
    }

    #[allow(clippy::too_many_lines)]
    #[instrument(skip_all, fields(i, nm, uuid = tracing::field::display(Uuid::from_bytes(uuid))
    ), err)]
    pub fn update_model(
        &self,
        model_idx: u8,
        uuid: [u8; 16],
        payload: &[u8],
        txn: Option<WriteTransaction<'_>>,
    ) -> anyhow::Result<()> {
        if let Some(model) = self.model_configs.get(model_idx as usize) {
            let curr_span = tracing::Span::current();
            curr_span.record("i", model.idx);
            curr_span.record("nm", tracing::field::display(&model.name));

            let root_vec = Reader::get_root(payload)?.as_vector();

            let mut builder = flexbuffers::Builder::new(&flexbuffers::BuilderOptions::SHARE_NONE);
            let mut item = builder.start_vector();

            item.push(flexbuffers::Blob(uuid.as_slice()));

            let mut index_key = BytesMut::new();

            index_key.put_u8(model_idx);
            index_key.put_u8(0);
            index_key.put(uuid.as_slice());

            let mut search = None;
            let mut has_search_field_changes = false;

            let txn = match txn {
                Some(txn) => txn,
                None => WriteTransaction::new(self.env.clone())?,
            };

            {
                let mut access = txn.access();

                let item_key = access.get::<[u8], [u8]>(&self.index_db, index_key.as_ref())?;
                let item_key = Bytes::copy_from_slice(item_key);

                if self.limits.search_enabled
                    && let Some(searchable) = self.model_search_indexes.get(model_idx as usize)
                    && let Some((schema, index, _reader)) = searchable
                {
                    let mut doc = TantivyDocument::default();
                    let id_field = schema.get_field("0")?;
                    doc.add_bytes(id_field, item_key.as_ref());

                    search = Some((
                        doc,
                        schema,
                        index,
                        Term::from_field_bytes(id_field, item_key.as_ref()),
                    ));
                }

                let existing_item = access.get::<[u8], [u8]>(&self.item_db, item_key.as_ref())?;
                let existing_item = Bytes::copy_from_slice(existing_item);

                let existing_root_vec = Reader::get_root(existing_item.as_ref())?.as_vector();

                for field in &model.fields {
                    let reader = root_vec.idx(field.idx as usize - 1);
                    let existing_reader = existing_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)) =
                            self.cross_refs.get(&(model_idx, field.idx))
                        {
                            // many-to-many
                            if many == &Some(true) && *ref_is_many {
                                let mut existing_ref_item_ids = HashSet::new();
                                let mut seen_ref_item_ids = HashSet::new();

                                for ref_reader in &existing_reader.as_vector() {
                                    let ref_item_id_bytes = ref_reader.as_blob().0;

                                    existing_ref_item_ids
                                        .insert(Bytes::copy_from_slice(ref_item_id_bytes));
                                }

                                let mut refs = item.start_vector();
                                let mut seen_uuids = HashSet::new();

                                for uuid in &reader.as_vector() {
                                    let uuid = uuid.as_blob();
                                    let uuid_bytes = uuid.0;
                                    let mut lookup = BytesMut::new();

                                    lookup.put_u8(*ref_model_idx);
                                    lookup.put_u8(0);

                                    lookup.put(uuid_bytes);

                                    if seen_uuids.contains(uuid_bytes) {
                                        continue;
                                    }
                                    seen_uuids.insert(Bytes::copy_from_slice(uuid_bytes));

                                    let ref_item_id = access
                                        .get::<[u8], [u8]>(&self.index_db, lookup.as_ref())?;
                                    let ref_item_id = Bytes::copy_from_slice(ref_item_id);

                                    refs.push(flexbuffers::Blob(ref_item_id.as_ref()));
                                    seen_ref_item_ids.insert(ref_item_id.clone());

                                    if existing_ref_item_ids.contains(&ref_item_id) {
                                        continue;
                                    }

                                    // update foreign
                                    let ref_item = access
                                        .get::<[u8], [u8]>(&self.item_db, ref_item_id.as_ref())?;

                                    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(uuid);

                                    for ref_field in ref_model_fields {
                                        if &ref_field.idx == ref_field_idx {
                                            let mut curr_refs = builder_vec.start_vector();

                                            // pushes self ref to front of foreign's refs
                                            curr_refs.push(flexbuffers::Blob(item_key.as_ref()));

                                            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(
                                        &self.item_db,
                                        ref_item_id.as_ref(),
                                        updated_ref_item,
                                        &put::Flags::empty(),
                                    )?;
                                }

                                let mut clean_ids = vec![];

                                for existing_id in existing_ref_item_ids {
                                    if !seen_ref_item_ids.contains(&existing_id) {
                                        clean_ids.push(existing_id);
                                    }
                                }

                                for clean_id in clean_ids {
                                    // update foreign
                                    let ref_item = access
                                        .get::<[u8], [u8]>(&self.item_db, clean_id.as_ref())?;

                                    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(
                                        &self.item_db,
                                        clean_id.as_ref(),
                                        updated_ref_item,
                                        &put::Flags::empty(),
                                    )?;
                                }

                                refs.end_vector();

                            // one-of-many (on another record)
                            } else if *ref_is_many {
                                let uuid = reader.as_blob();
                                let uuid_bytes = uuid.0;
                                let mut lookup = BytesMut::new();

                                lookup.put_u8(*ref_model_idx);
                                lookup.put_u8(0);

                                lookup.put(uuid_bytes);

                                let ref_item_id =
                                    access.get::<[u8], [u8]>(&self.index_db, lookup.as_ref())?;

                                let existing_ref_item_id = existing_reader.as_blob();
                                let existing_ref_item_id_bytes = existing_ref_item_id.0;

                                if ref_item_id == existing_ref_item_id_bytes {
                                    item.push(existing_ref_item_id);
                                } else {
                                    let ref_item_id = Bytes::copy_from_slice(ref_item_id);

                                    let existing_ref_item = access.get::<[u8], [u8]>(
                                        &self.item_db,
                                        existing_ref_item_id_bytes,
                                    )?;
                                    let ref_item = access
                                        .get::<[u8], [u8]>(&self.item_db, ref_item_id.as_ref())?;

                                    let existing_root_vec =
                                        Reader::get_root(existing_ref_item)?.as_vector();
                                    let mut existing_builder = flexbuffers::Builder::new(
                                        &flexbuffers::BuilderOptions::SHARE_NONE,
                                    );
                                    let mut existing_builder_vec = existing_builder.start_vector();
                                    existing_builder_vec.push(existing_root_vec.idx(0).as_blob());

                                    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(uuid);

                                    for ref_field in ref_model_fields {
                                        if &ref_field.idx == ref_field_idx {
                                            let mut existing_curr_refs =
                                                existing_builder_vec.start_vector();

                                            for curr_ref in &existing_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 {
                                                    existing_curr_refs.push(curr_ref);
                                                }
                                            }

                                            existing_curr_refs.end_vector();

                                            let mut curr_refs = builder_vec.start_vector();

                                            // pushes self ref to front of foreign's refs
                                            curr_refs.push(flexbuffers::Blob(item_key.as_ref()));

                                            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,
                                            )?;
                                            ref_field.kind.copy_to(
                                                &existing_root_vec.idx(ref_field.idx as usize),
                                                &mut existing_builder_vec,
                                                None,
                                            )?;
                                        }
                                    }

                                    builder_vec.end_vector();
                                    existing_builder_vec.end_vector();

                                    let updated_ref_item = builder.view();
                                    let updated_existing_ref_item = existing_builder.view();

                                    access.put(
                                        &self.item_db,
                                        existing_ref_item_id_bytes,
                                        updated_existing_ref_item,
                                        &put::Flags::empty(),
                                    )?;
                                    access.put(
                                        &self.item_db,
                                        ref_item_id.as_ref(),
                                        updated_ref_item,
                                        &put::Flags::empty(),
                                    )?;
                                }

                            // many-to-one
                            } else {
                                // preserve existing as children dictate / self-assign to parent

                                let mut ref_ids = item.start_vector();

                                for ref_id in &existing_reader.as_vector() {
                                    let ref_id = ref_id.as_blob();
                                    ref_ids.push(ref_id);
                                }

                                ref_ids.end_vector();
                            }
                        }
                    } else {
                        if field.indexed == Some(true) {
                            let mut key = BytesMut::new();
                            key.put_u8(model_idx);
                            key.put_u8(field.idx);

                            let mut existing_key = BytesMut::new();
                            key.put_u8(model_idx);
                            key.put_u8(field.idx);

                            match field.kind {
                                Kind::String | Kind::Url => {
                                    key.put(reader.as_str().as_bytes());
                                    existing_key.put(existing_reader.as_str().as_bytes());
                                }
                                Kind::Uuid => {
                                    key.put(reader.as_blob().0);
                                    existing_key.put(existing_reader.as_blob().0);
                                }
                                _ => {
                                    bail!("cannot index on {:?}", field.kind);
                                }
                            }

                            if key != existing_key {
                                if access
                                    .get::<[u8], [u8]>(&self.index_db, key.as_ref())
                                    .is_ok()
                                {
                                    bail!("value has already been indexed");
                                }

                                access.put(
                                    &self.index_db,
                                    key.as_ref(),
                                    item_key.as_ref(),
                                    &put::Flags::empty(),
                                )?;

                                access.del_key(&self.index_db, existing_key.as_ref())?;
                            }
                        }

                        if field.queryable == Some(true) {
                            let mut key = BytesMut::new();
                            key.put_u8(model_idx);
                            key.put_u8(field.idx);

                            let mut existing_key = BytesMut::new();
                            key.put_u8(model_idx);
                            key.put_u8(field.idx);

                            match &field.kind {
                                Kind::Bool => {
                                    if reader.as_bool() {
                                        key.put_u8(1);
                                    } else {
                                        key.put_u8(0);
                                    }

                                    if existing_reader.as_bool() {
                                        existing_key.put_u8(1);
                                    } else {
                                        existing_key.put_u8(0);
                                    }
                                }
                                Kind::String | Kind::Url => {
                                    key.put(reader.as_str().as_bytes());
                                    existing_key.put(existing_reader.as_str().as_bytes());
                                }
                                Kind::Uuid => {
                                    key.put(reader.as_blob().0);
                                    existing_key.put(existing_reader.as_blob().0);
                                }
                                Kind::Timestamp { unit } => match &unit {
                                    TimeUnit::Seconds => {
                                        key.put_i64(reader.as_i64());
                                        existing_key.put_i64(existing_reader.as_i64());
                                    }
                                },
                                Kind::U8 => {
                                    key.put_u8(reader.as_u8());
                                    existing_key.put_u8(existing_reader.as_u8());
                                }
                                Kind::U16 => {
                                    key.put_u16(reader.as_u16());
                                    existing_key.put_u16(existing_reader.as_u16());
                                }
                                Kind::U32 => {
                                    key.put_u32(reader.as_u32());
                                    existing_key.put_u32(existing_reader.as_u32());
                                }
                                Kind::U64 => {
                                    key.put_u64(reader.as_u64());
                                    existing_key.put_u64(existing_reader.as_u64());
                                }
                                Kind::I8 => {
                                    key.put_i8(reader.as_i8());
                                    existing_key.put_i8(existing_reader.as_i8());
                                }
                                Kind::I16 => {
                                    key.put_i16(reader.as_i16());
                                    existing_key.put_i16(existing_reader.as_i16());
                                }
                                Kind::I32 => {
                                    key.put_i32(reader.as_i32());
                                    existing_key.put_i32(existing_reader.as_i32());
                                }
                                Kind::I64 => {
                                    key.put_i64(reader.as_i64());
                                    existing_key.put_i64(existing_reader.as_i64());
                                }
                                Kind::F32 => {
                                    key.put_f32(reader.as_f32());
                                    existing_key.put_f32(existing_reader.as_f32());
                                }
                                Kind::F64 => {
                                    key.put_f64(reader.as_f64());
                                    existing_key.put_f64(existing_reader.as_f64());
                                }
                                _ => {
                                    bail!("cannot query on {:?}", field.kind);
                                }
                            }

                            if key != existing_key {
                                // remove existing
                                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]>(&self.query_db, existing_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(
                                    &self.query_db,
                                    existing_key.as_ref(),
                                    updated_list,
                                    &put::Flags::empty(),
                                )?;

                                // add to new
                                let mut builder = flexbuffers::Builder::new(
                                    &flexbuffers::BuilderOptions::SHARE_NONE,
                                );
                                let mut builder_vec = builder.start_vector();

                                builder_vec.push(flexbuffers::Blob(item_key.as_ref()));

                                if let Ok(list) =
                                    access.get::<[u8], [u8]>(&self.query_db, key.as_ref())
                                {
                                    let root = Reader::get_root(list)?;
                                    for val in &root.as_vector() {
                                        builder_vec.push(val.as_blob());
                                    }
                                }

                                builder_vec.end_vector();

                                let updated_list = builder.view();

                                access.put(
                                    &self.query_db,
                                    key.as_ref(),
                                    updated_list,
                                    &put::Flags::empty(),
                                )?;
                            }
                        }

                        if field.searchable == Some(true) {
                            if let Some((doc, schema, _index, _term)) = &mut search {
                                let tantivy_field = schema.get_field(&field.idx.to_string())?;
                                add_field_to_doc_flexbuffer(doc, field, tantivy_field, &reader);
                            }

                            if field_to_bytes(field, &reader)
                                != field_to_bytes(field, &existing_reader)
                            {
                                has_search_field_changes = true;
                            }
                        }

                        self.encrypt_compress_copy_to(field, &reader, &mut item)?;
                    }
                }

                item.end_vector();
                let item = builder.view();

                if item.len() as u64 > self.limits.max_item_size {
                    bail!("item size exceeds limits");
                }

                access.put(&self.item_db, item_key.as_ref(), item, &put::Flags::empty())?;
            }

            txn.commit()?;

            if has_search_field_changes && let Some((doc, _schema, index, del_term)) = search {
                // todo: see if this is reusable so it doesn't have to be allocated on every write
                let mut index_writer: IndexWriter = index.writer(15_000_000)?;

                index_writer.delete_term(del_term);
                index_writer.add_document(doc)?;

                index_writer.commit()?;
                index_writer.wait_merging_threads()?;
            }
        }

        Ok(())
    }

    #[allow(clippy::too_many_lines)]
    fn build_item(
        &self,
        model_config: &ModelConfig,
        access: &Accessor,
        item_key: &[u8],
        item_vec: &mut flexbuffers::VectorBuilder,
        ref_depth: &RefDepth,
        mut count: usize,
    ) -> anyhow::Result<usize> {
        let item = match access {
            Accessor::Write(access) => access.get::<[u8], [u8]>(&self.item_db, item_key)?,
            Accessor::Const(access) => access.get::<[u8], [u8]>(&self.item_db, item_key)?,
        };

        let root_vec = Reader::get_root(item)?.as_vector();

        let item_uuid = root_vec.idx(0).as_blob();
        item_vec.push(item_uuid);

        for field in &model_config.fields {
            match &field.kind {
                Kind::Ref {
                    model,
                    field: _,
                    many,
                } => {
                    if let Some(((field_idx, limit, cursor), ref_depth)) =
                        ref_depth.0.iter().find(|v| v.0.0 == field.idx)
                    {
                        if let Some(model_config) = self.model_config_map.get(model) {
                            if many == &Some(true) {
                                let mut skip_key = match cursor {
                                    Some(cursor) => {
                                        let mut index_key = BytesMut::new();

                                        index_key.put_u8(model_config.idx);
                                        index_key.put_u8(0);
                                        index_key.put(cursor.as_ref());

                                        let item_key = match access {
                                            Accessor::Write(access) => access.get::<[u8], [u8]>(
                                                &self.index_db,
                                                index_key.as_ref(),
                                            )?,
                                            Accessor::Const(access) => access.get::<[u8], [u8]>(
                                                &self.index_db,
                                                index_key.as_ref(),
                                            )?,
                                        };
                                        Some(item_key)
                                    }
                                    None => None,
                                };

                                let mut limit = *limit as usize;
                                let mut ref_list_vec = item_vec.start_vector();

                                for ref_id in &root_vec.idx(*field_idx as usize).as_vector() {
                                    if limit == 0 {
                                        break;
                                    }

                                    let ref_id_bytes = ref_id.as_blob().0;

                                    if let Some(sk) = skip_key {
                                        if sk == ref_id_bytes {
                                            skip_key = None;
                                        }

                                        continue;
                                    }

                                    let mut ref_vec = ref_list_vec.start_vector();

                                    count = self.build_item(
                                        model_config,
                                        access,
                                        ref_id_bytes,
                                        &mut ref_vec,
                                        ref_depth,
                                        count,
                                    )?;

                                    ref_vec.end_vector();

                                    limit -= 1;
                                }

                                ref_list_vec.end_vector();
                            } else if *limit > 0 {
                                let ref_id_bytes = root_vec.idx(*field_idx as usize).as_blob().0;

                                let mut ref_vec = item_vec.start_vector();

                                count = self.build_item(
                                    model_config,
                                    access,
                                    ref_id_bytes,
                                    &mut ref_vec,
                                    ref_depth,
                                    count,
                                )?;

                                ref_vec.end_vector();
                            } else {
                                item_vec.push(());
                            }
                        }
                    } else {
                        item_vec.push(());
                    }
                }
                _ => {
                    if field.encrypted == Some(true) {
                        let cipher = XChaCha20Poly1305::new(&self.encryption_key.into());
                        let bytes = root_vec.idx(field.idx as usize).as_blob().0;
                        let len = bytes.len() - 24;

                        let ciphertext = &bytes[..len - 24];
                        let nonce: [u8; 24] = bytes[len - 24..].try_into()?;
                        let nonce = XNonce::try_from(nonce.as_slice())?;

                        match cipher.decrypt(&nonce, ciphertext) {
                            Ok(plaintext) => {
                                if field.compressed == Some(true) {
                                    let decompressed =
                                        zstd::stream::decode_all(plaintext.as_slice())?;
                                    push_field_from_bytes(field, &decompressed, item_vec)?;
                                } else {
                                    push_field_from_bytes(field, &plaintext, item_vec)?;
                                }
                            }
                            Err(err) => tracing::error!(%err),
                        }
                    } else if field.compressed == Some(true) {
                        let bytes = root_vec.idx(field.idx as usize).as_blob().0;

                        let decompressed = zstd::stream::decode_all(bytes)?;

                        push_field_from_bytes(field, &decompressed, item_vec)?;
                    } else {
                        field
                            .kind
                            .copy_to(&root_vec.idx(field.idx as usize), item_vec, None)?;
                    }
                }
            }
        }

        Ok(count + 1)
    }

    #[instrument(skip_all, fields(i, nm), err)]
    pub fn get_model(
        &self,
        model_idx: u8,
        field_idx: u8,
        index_val: &[u8],
        ref_depth: &RefDepth,
        txn: Option<WriteTransaction<'_>>,
    ) -> anyhow::Result<Bytes> {
        if let Some(model_config) = self.model_configs.get(model_idx as usize) {
            let curr_span = tracing::Span::current();
            curr_span.record("i", model_config.idx);
            curr_span.record("nm", tracing::field::display(&model_config.name));

            if let Some(idx) = (field_idx as usize).checked_sub(1) {
                if let Some(field) = model_config.fields.get(idx) {
                    match field.kind {
                        Kind::String | Kind::Url => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            v = %std::str::from_utf8(index_val)?
                        ),
                        Kind::Uuid => {
                            if index_val.len() == 16 {
                                let uuid: [u8; 16] = index_val[0..16].try_into()?;
                                tracing::info!(
                                    f = field.idx,
                                    nm = %field.name,
                                    v = %Uuid::from_bytes(uuid)
                                );
                            } else {
                                bail!("uuid is not 16 bytes");
                            }
                        }
                        _ => {
                            tracing::error!("kind {:?} not yet supported as indexable", field.kind);
                        }
                    }
                } else {
                    bail!("no field for idx {field_idx} on model {model_idx}");
                }
            } else if index_val.len() == 16 {
                let uuid: [u8; 16] = index_val[0..16].try_into()?;
                tracing::info!(f = 0, nm = %"uuid", v = %Uuid::from_bytes(uuid));
            } else {
                bail!("uuid is not 16 bytes");
            }

            let mut index_key = BytesMut::new();

            index_key.put_u8(model_idx);
            index_key.put_u8(field_idx);
            index_key.put(index_val);

            let mut result_builder =
                flexbuffers::Builder::new(&flexbuffers::BuilderOptions::SHARE_NONE);
            let mut item_vec = result_builder.start_vector();

            let txn = match &txn {
                Some(txn) => Transaction::Write(txn),
                None => Transaction::Read(&ReadTransaction::new(self.env.clone())?),
            };

            let count = match txn {
                Transaction::Write(txn) => {
                    let access = txn.access();

                    let item_key = access.get::<[u8], [u8]>(&self.index_db, index_key.as_ref())?;

                    self.build_item(
                        model_config,
                        &Accessor::Write(&access),
                        item_key,
                        &mut item_vec,
                        ref_depth,
                        0,
                    )?
                }
                Transaction::Read(txn) => {
                    let access = txn.access();

                    let item_key = access.get::<[u8], [u8]>(&self.index_db, index_key.as_ref())?;

                    self.build_item(
                        model_config,
                        &Accessor::Const(&access),
                        item_key,
                        &mut item_vec,
                        ref_depth,
                        0,
                    )?
                }
            };

            tracing::info!(ct = count);
            item_vec.end_vector();

            Ok(Bytes::copy_from_slice(result_builder.view()))
        } else {
            bail!("no model for idx {model_idx}")
        }
    }

    fn get_for_keys(
        &self,
        model_config: &ModelConfig,
        access: &Accessor,
        keys: &[u8],
        result: &mut flexbuffers::VectorBuilder,
        ref_depth: &RefDepth,
        mut count: usize,
    ) -> anyhow::Result<usize> {
        let root = Reader::get_root(keys)?;

        for key in &root.as_vector() {
            let item_key_bytes = key.as_blob().0;

            let mut item_vec = result.start_vector();
            count = self.build_item(
                model_config,
                access,
                item_key_bytes,
                &mut item_vec,
                ref_depth,
                count,
            )?;

            item_vec.end_vector();
        }

        Ok(count)
    }

    #[allow(clippy::too_many_lines)]
    #[instrument(
        skip_all
        fields(i, nm),
        err
    )]
    pub fn query_model(
        &self,
        model_idx: u8,
        field_idx: u8,
        expression: u8,
        value: &[u8],
        ref_depth: &RefDepth,
        txn: Option<WriteTransaction<'_>>,
    ) -> anyhow::Result<Bytes> {
        if let Some(model_config) = self.model_configs.get(model_idx as usize) {
            let curr_span = tracing::Span::current();
            curr_span.record("i", model_config.idx);
            curr_span.record("nm", tracing::field::display(&model_config.name));

            let exp = match expression {
                0 => "gte",
                1 => "lte",
                2 => "eq",
                3 => "gt",
                4 => "lt",
                5 => "begins_with",
                _ => bail!("invalid expression {expression}"),
            };

            if let Some(idx) = (field_idx as usize).checked_sub(1) {
                if let Some(field) = model_config.fields.get(idx) {
                    match &field.kind {
                        Kind::Bool => {
                            if value.len() != 1 {
                                bail!("invalid boolean");
                            }
                            tracing::info!(f = field.idx, nm = %field.name, exp, v = value[0] == 1,);
                        }
                        Kind::String | Kind::Url => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            exp,
                            v = %std::str::from_utf8(value)?
                        ),
                        Kind::Uuid => {
                            if value.len() == 16 {
                                let uuid: [u8; 16] = value[0..16].try_into()?;
                                tracing::info!(
                                    f = field.idx,
                                    nm = %field.name,
                                    exp,
                                    v = %Uuid::from_bytes(uuid)
                                );
                            } else {
                                bail!("uuid value is not 16 bytes");
                            }
                        }
                        Kind::Timestamp { unit } => match &unit {
                            TimeUnit::Seconds => {
                                tracing::info!(
                                    f = field.idx,
                                    nm = %field.name,
                                    exp,
                                    v = i64::from_be_bytes(value.as_ref().try_into()?)
                                );
                            }
                        },
                        Kind::I8 => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            exp,
                            v = i8::from_be_bytes(value.as_ref().try_into()?)
                        ),
                        Kind::I16 => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            exp,
                            v = i16::from_be_bytes(value.as_ref().try_into()?)
                        ),
                        Kind::I32 => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            exp,
                            v = i32::from_be_bytes(value.as_ref().try_into()?)
                        ),
                        Kind::I64 => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            exp,
                            v = i64::from_be_bytes(value.as_ref().try_into()?)
                        ),
                        Kind::U8 => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            exp,
                            v = u8::from_be_bytes(value.as_ref().try_into()?)
                        ),
                        Kind::U16 => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            exp,
                            v = u16::from_be_bytes(value.as_ref().try_into()?)
                        ),
                        Kind::U32 => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            exp,
                            v = u32::from_be_bytes(value.as_ref().try_into()?)
                        ),
                        Kind::U64 => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            exp,
                            v = u64::from_be_bytes(value.as_ref().try_into()?)
                        ),
                        Kind::F32 => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            exp,
                            v = f32::from_be_bytes(value.as_ref().try_into()?)
                        ),
                        Kind::F64 => tracing::info!(
                            f = field.idx,
                            nm = %field.name,
                            exp,
                            v = f64::from_be_bytes(value.as_ref().try_into()?)
                        ),
                        _ => {
                            tracing::error!("kind {:?} not yet supported as queryable", field.kind);
                        }
                    }
                } else {
                    bail!("no field for idx {field_idx} on model {model_idx}");
                }
            } else if value.len() == 16 {
                let uuid: [u8; 16] = value[0..16].try_into()?;
                tracing::info!(f = 0, nm = %"uuid", exp, v = %Uuid::from_bytes(uuid));
            } else {
                bail!("uuid value is not 16 bytes");
            }

            let mut query_key = BytesMut::new();

            query_key.put_u8(model_idx);
            query_key.put_u8(field_idx);
            query_key.put(value);

            let mut count = 0;

            let mut builder = flexbuffers::Builder::new(&flexbuffers::BuilderOptions::SHARE_NONE);
            let mut result = builder.start_vector();

            let txn = match &txn {
                Some(txn) => Transaction::Write(txn),
                None => Transaction::Read(&ReadTransaction::new(self.env.clone())?),
            };

            match expression {
                // gte / gt
                0 | 3 => match txn {
                    Transaction::Write(txn) => {
                        let access = txn.access();

                        let mut cursor = txn.cursor(self.index_db.clone())?;
                        let (qk, keys) = cursor.seek_range_k(&access, query_key.as_ref())?;

                        if (expression == 0 || qk != query_key.as_ref())
                            && qk[0] == model_idx
                            && qk[1] == field_idx
                        {
                            count = self.get_for_keys(
                                model_config,
                                &Accessor::Write(&access),
                                keys,
                                &mut result,
                                ref_depth,
                                count,
                            )?;
                        }

                        loop {
                            if let Ok((qk, keys)) = cursor.next::<[u8], [u8]>(&access)
                                && qk[0] == model_idx
                                && qk[1] == field_idx
                            {
                                count = self.get_for_keys(
                                    model_config,
                                    &Accessor::Write(&access),
                                    keys,
                                    &mut result,
                                    ref_depth,
                                    count,
                                )?;
                            }
                        }
                    }
                    Transaction::Read(txn) => {
                        let access = txn.access();

                        let mut cursor = txn.cursor(self.index_db.clone())?;
                        let (qk, keys) = cursor.seek_range_k(&access, query_key.as_ref())?;

                        if (expression == 0 || qk != query_key.as_ref())
                            && qk[0] == model_idx
                            && qk[1] == field_idx
                        {
                            count = self.get_for_keys(
                                model_config,
                                &Accessor::Const(&access),
                                keys,
                                &mut result,
                                ref_depth,
                                count,
                            )?;
                        }

                        loop {
                            if let Ok((qk, keys)) = cursor.next::<[u8], [u8]>(&access)
                                && qk[0] == model_idx
                                && qk[1] == field_idx
                            {
                                count = self.get_for_keys(
                                    model_config,
                                    &Accessor::Const(&access),
                                    keys,
                                    &mut result,
                                    ref_depth,
                                    count,
                                )?;
                            }
                        }
                    }
                },
                // lte / lt
                1 | 4 => match txn {
                    Transaction::Write(txn) => {
                        let access = txn.access();

                        let mut cursor = txn.cursor(self.index_db.clone())?;
                        let (qk, keys) = cursor.seek_range_k(&access, query_key.as_ref())?;

                        if expression == 1 && qk == query_key.as_ref() {
                            count = self.get_for_keys(
                                model_config,
                                &Accessor::Write(&access),
                                keys,
                                &mut result,
                                ref_depth,
                                count,
                            )?;
                        }

                        loop {
                            if let Ok((qk, keys)) = cursor.prev::<[u8], [u8]>(&access)
                                && qk[0] == model_idx
                                && qk[1] == field_idx
                            {
                                count = self.get_for_keys(
                                    model_config,
                                    &Accessor::Write(&access),
                                    keys,
                                    &mut result,
                                    ref_depth,
                                    count,
                                )?;
                            }
                        }
                    }
                    Transaction::Read(txn) => {
                        let access = txn.access();

                        let mut cursor = txn.cursor(self.index_db.clone())?;
                        let (qk, keys) = cursor.seek_range_k(&access, query_key.as_ref())?;

                        if expression == 1 && qk == query_key.as_ref() {
                            count = self.get_for_keys(
                                model_config,
                                &Accessor::Const(&access),
                                keys,
                                &mut result,
                                ref_depth,
                                count,
                            )?;
                        }

                        loop {
                            if let Ok((qk, keys)) = cursor.prev::<[u8], [u8]>(&access)
                                && qk[0] == model_idx
                                && qk[1] == field_idx
                            {
                                count = self.get_for_keys(
                                    model_config,
                                    &Accessor::Const(&access),
                                    keys,
                                    &mut result,
                                    ref_depth,
                                    count,
                                )?;
                            }
                        }
                    }
                },
                // eq
                2 => match txn {
                    Transaction::Write(txn) => {
                        let access = txn.access();
                        if let Ok(keys) =
                            access.get::<[u8], [u8]>(&self.query_db, query_key.as_ref())
                        {
                            count = self.get_for_keys(
                                model_config,
                                &Accessor::Write(&access),
                                keys,
                                &mut result,
                                ref_depth,
                                count,
                            )?;
                        }
                    }
                    Transaction::Read(txn) => {
                        let access = txn.access();
                        if let Ok(keys) =
                            access.get::<[u8], [u8]>(&self.query_db, query_key.as_ref())
                        {
                            count = self.get_for_keys(
                                model_config,
                                &Accessor::Const(&access),
                                keys,
                                &mut result,
                                ref_depth,
                                count,
                            )?;
                        }
                    }
                },
                // begins_with
                5 => {
                    let query_key_len = query_key.len();

                    match txn {
                        Transaction::Write(txn) => {
                            let access = txn.access();

                            let mut cursor = txn.cursor(self.index_db.clone())?;
                            let (qk, keys) = cursor.seek_range_k(&access, query_key.as_ref())?;

                            if qk[0] == model_idx
                                && qk[1] == field_idx
                                && qk.len() >= query_key_len
                                && qk[..query_key_len] == query_key
                            {
                                count = self.get_for_keys(
                                    model_config,
                                    &Accessor::Write(&access),
                                    keys,
                                    &mut result,
                                    ref_depth,
                                    count,
                                )?;
                            }

                            loop {
                                if let Ok((qk, keys)) = cursor.next::<[u8], [u8]>(&access)
                                    && qk[0] == model_idx
                                    && qk[1] == field_idx
                                    && qk.len() >= query_key_len
                                    && qk[..query_key_len] == query_key
                                {
                                    count = self.get_for_keys(
                                        model_config,
                                        &Accessor::Write(&access),
                                        keys,
                                        &mut result,
                                        ref_depth,
                                        count,
                                    )?;
                                }
                            }
                        }
                        Transaction::Read(txn) => {
                            let access = txn.access();

                            let mut cursor = txn.cursor(self.index_db.clone())?;
                            let (qk, keys) = cursor.seek_range_k(&access, query_key.as_ref())?;

                            if qk[0] == model_idx
                                && qk[1] == field_idx
                                && qk.len() >= query_key_len
                                && qk[..query_key_len] == query_key
                            {
                                count = self.get_for_keys(
                                    model_config,
                                    &Accessor::Const(&access),
                                    keys,
                                    &mut result,
                                    ref_depth,
                                    count,
                                )?;
                            }

                            loop {
                                if let Ok((qk, keys)) = cursor.next::<[u8], [u8]>(&access)
                                    && qk[0] == model_idx
                                    && qk[1] == field_idx
                                    && qk.len() >= query_key_len
                                    && qk[..query_key_len] == query_key
                                {
                                    count = self.get_for_keys(
                                        model_config,
                                        &Accessor::Const(&access),
                                        keys,
                                        &mut result,
                                        ref_depth,
                                        count,
                                    )?;
                                }
                            }
                        }
                    }
                }
                _ => {}
            }

            result.end_vector();
            tracing::info!(ct = count);

            Ok(Bytes::copy_from_slice(builder.view()))
        } else {
            bail!("no model for idx {model_idx}")
        }
    }

    #[allow(clippy::too_many_lines)]
    #[instrument(skip_all, fields(i, nm), err)]
    pub fn search_model(
        &self,
        model_idx: u8,
        field_idx: u8,
        value: &[u8],
        ref_depth: &RefDepth,
        txn: Option<WriteTransaction<'_>>,
    ) -> anyhow::Result<Bytes> {
        if let Some(model_config) = self.model_configs.get(model_idx as usize) {
            let curr_span = tracing::Span::current();
            curr_span.record("i", model_config.idx);
            curr_span.record("nm", tracing::field::display(&model_config.name));

            let mut count = 0;

            if let Some(searchable) = self.model_search_indexes.get(model_idx as usize)
                && let Some((schema, _index, reader)) = searchable
            {
                let field_kind: Kind;

                if let Some(idx) = (field_idx as usize).checked_sub(1) {
                    if let Some(field) = model_config.fields.get(idx) {
                        match &field.kind {
                            Kind::Bool => {
                                if value.len() != 1 {
                                    bail!("invalid boolean");
                                }
                                tracing::info!(f = field.idx, nm = %field.name, v = value[0] == 1,);
                            }
                            Kind::String | Kind::Url | Kind::Json | Kind::Markdown => {
                                tracing::info!(
                                    f = field.idx,
                                    nm = %field.name,
                                    v = std::str::from_utf8(value)?
                                );
                            }
                            Kind::Uuid => {
                                if value.len() == 16 {
                                    let uuid: [u8; 16] = value[0..16].try_into()?;
                                    tracing::info!(
                                        f = field.idx,
                                        nm = %field.name,
                                        v = %Uuid::from_bytes(uuid)
                                    );
                                } else {
                                    bail!("uuid value is not 16 bytes");
                                }
                            }
                            Kind::Timestamp { unit } => match &unit {
                                TimeUnit::Seconds => {
                                    tracing::info!(
                                        f = field.idx,
                                        nm = %field.name,
                                        v = i64::from_be_bytes(value.as_ref().try_into()?)
                                    );
                                }
                            },
                            Kind::I8 => tracing::info!(
                                f = field.idx,
                                nm = %field.name,
                                v = i8::from_be_bytes(value.as_ref().try_into()?)
                            ),
                            Kind::I16 => tracing::info!(
                                f = field.idx,
                                nm = %field.name,
                                v = i16::from_be_bytes(value.as_ref().try_into()?)
                            ),
                            Kind::I32 => tracing::info!(
                                f = field.idx,
                                nm = %field.name,
                                v = i32::from_be_bytes(value.as_ref().try_into()?)
                            ),
                            Kind::I64 => tracing::info!(
                                f = field.idx,
                                nm = %field.name,
                                v = i64::from_be_bytes(value.as_ref().try_into()?)
                            ),
                            Kind::U8 => tracing::info!(
                                f = field.idx,
                                nm = %field.name,
                                v = u8::from_be_bytes(value.as_ref().try_into()?)
                            ),
                            Kind::U16 => tracing::info!(
                                f = field.idx,
                                nm = %field.name,
                                v = u16::from_be_bytes(value.as_ref().try_into()?)
                            ),
                            Kind::U32 => tracing::info!(
                                f = field.idx,
                                nm = %field.name,
                                v = u32::from_be_bytes(value.as_ref().try_into()?)
                            ),
                            Kind::U64 => tracing::info!(
                                f = field.idx,
                                nm = %field.name,
                                v = u64::from_be_bytes(value.as_ref().try_into()?)
                            ),
                            Kind::F32 => tracing::info!(
                                f = field.idx,
                                nm = %field.name,
                                v = f32::from_be_bytes(value.as_ref().try_into()?)
                            ),
                            Kind::F64 => tracing::info!(
                                f = field.idx,
                                nm = %field.name,
                                v = f64::from_be_bytes(value.as_ref().try_into()?)
                            ),
                            _ => {
                                tracing::error!(
                                    "kind {:?} not yet supported as queryable",
                                    field.kind
                                );
                            }
                        }
                        field_kind = field.kind.clone();
                    } else {
                        bail!("no field idx {field_idx} for model {model_idx}");
                    }
                } else {
                    field_kind = Kind::Uuid;
                }

                let field = schema.get_field(&field_idx.to_string())?;

                let term = get_term(&field_kind, field, value)?;
                let query = FuzzyTermQuery::new(term, 2, true);

                let searcher = reader.searcher();
                let docs = searcher.search(&query, &TopDocs::with_limit(255).order_by_score())?;

                tracing::info!(docs = docs.len());

                let mut results_builder =
                    flexbuffers::Builder::new(&flexbuffers::BuilderOptions::SHARE_NONE);
                let mut results_vec = results_builder.start_vector();

                let txn = match &txn {
                    Some(txn) => Transaction::Write(txn),
                    None => Transaction::Read(&ReadTransaction::new(self.env.clone())?),
                };

                match txn {
                    Transaction::Write(txn) => {
                        let access = txn.access();

                        for (_score, doc_addr) in docs {
                            let doc: TantivyDocument = searcher.doc(doc_addr)?;
                            if let Some(item_key) = doc.get_first(schema.get_field("0")?)
                                && let Some(item_key) = item_key.as_bytes()
                            {
                                let mut item_vec = results_vec.start_vector();

                                count = self.build_item(
                                    model_config,
                                    &Accessor::Write(&access),
                                    item_key,
                                    &mut item_vec,
                                    ref_depth,
                                    count,
                                )?;
                                item_vec.end_vector();
                            }
                        }
                    }
                    Transaction::Read(txn) => {
                        let access = txn.access();

                        for (_score, doc_addr) in docs {
                            let doc: TantivyDocument = searcher.doc(doc_addr)?;
                            if let Some(item_key) = doc.get_first(schema.get_field("0")?)
                                && let Some(item_key) = item_key.as_bytes()
                            {
                                let mut item_vec = results_vec.start_vector();

                                count = self.build_item(
                                    model_config,
                                    &Accessor::Const(&access),
                                    item_key,
                                    &mut item_vec,
                                    ref_depth,
                                    count,
                                )?;
                                item_vec.end_vector();
                            }
                        }
                    }
                }

                results_vec.end_vector();
                tracing::info!(ct = count);

                return Ok(Bytes::copy_from_slice(results_builder.view()));
            }
            bail!("no searchable index for idx {model_idx}");
        }
        bail!("no model config for idx {model_idx}")
    }

    /// !! inefficient and only useful for admin purposes
    #[instrument(skip_all, fields(i, nm), err)]
    pub fn list_model(&self, model_idx: u8, cursor: Option<&[u8]>) -> anyhow::Result<Bytes> {
        if let Some(model_config) = self.model_configs.get(model_idx as usize) {
            let curr_span = tracing::Span::current();
            curr_span.record("i", model_config.idx);
            curr_span.record("nm", tracing::field::display(&model_config.name));

            let mut index_key = BytesMut::new();

            index_key.put_u8(model_idx);
            index_key.put_u8(0);

            if let Some(cursor) = cursor {
                index_key.put(cursor);
            }

            let mut count = 0;

            let mut builder = flexbuffers::Builder::new(&flexbuffers::BuilderOptions::SHARE_NONE);
            let mut result = builder.start_vector();

            let txn = ReadTransaction::new(self.env.clone())?;
            let access = txn.access();

            let mut cursor = txn.cursor(self.index_db.clone())?;

            let (k, v) = cursor.seek_range_k(&access, index_key.as_ref())?;

            if k[0] == model_idx && k[1] == 0 {
                let mut first_item_vec = result.start_vector();
                count = self.build_item(
                    model_config,
                    &Accessor::Const(&access),
                    v,
                    &mut first_item_vec,
                    &RefDepth(vec![]),
                    count,
                )?;
                first_item_vec.end_vector();
            }

            while let Ok((k, v)) = cursor.next::<[u8], [u8]>(&access) {
                if k[1] != 0 {
                    break;
                }

                if k[0] == model_idx {
                    let mut item_vec = result.start_vector();
                    count = self.build_item(
                        model_config,
                        &Accessor::Const(&access),
                        v,
                        &mut item_vec,
                        &RefDepth(vec![]),
                        count,
                    )?;
                    item_vec.end_vector();
                } else {
                    break;
                }
            }

            result.end_vector();
            tracing::info!(ct = count);

            Ok(Bytes::copy_from_slice(builder.view()))
        } else {
            bail!("no model with idx {model_idx}")
        }
    }

    #[allow(clippy::too_many_lines)]
    #[instrument(skip_all, fields(i, nm, uuid = tracing::field::display(Uuid::from_bytes(uuid))
    ), err)]
    pub fn delete_model(
        &self,
        model_idx: u8,
        uuid: [u8; 16],
        txn: Option<WriteTransaction<'_>>,
    ) -> anyhow::Result<()> {
        if let Some(model) = self.model_configs.get(model_idx as usize) {
            let curr_span = tracing::Span::current();
            curr_span.record("i", model.idx);
            curr_span.record("nm", tracing::field::display(&model.name));

            let mut index_key = BytesMut::new();

            index_key.put_u8(model_idx);
            index_key.put_u8(0);
            index_key.put(uuid.as_slice());

            let txn = match txn {
                Some(txn) => txn,
                None => WriteTransaction::new(self.env.clone())?,
            };

            {
                let mut access = txn.access();

                let item_key = access.get::<[u8], [u8]>(&self.index_db, &index_key)?;
                let item = access.get::<[u8], [u8]>(&self.item_db, item_key)?;

                let item_key = Bytes::copy_from_slice(item_key);
                let item = Bytes::copy_from_slice(item);

                let root_vec = Reader::get_root(item.as_ref())?.as_vector();

                for field in &model.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,
                        )) = self.cross_refs.get(&(model_idx, field.idx))
                        {
                            // many-to-many
                            if many == &Some(true) && *ref_is_many {
                                for ref_reader in &reader.as_vector() {
                                    let ref_item_id_bytes = ref_reader.as_blob().0;

                                    // update foreign
                                    let ref_item = access
                                        .get::<[u8], [u8]>(&self.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(
                                        &self.item_db,
                                        ref_item_id_bytes,
                                        updated_ref_item,
                                        &put::Flags::empty(),
                                    )?;
                                }

                            // one-of-many (on another record)
                            } else if *ref_is_many {
                                let ref_item_id_bytes = reader.as_blob().0;

                                let ref_item =
                                    access.get::<[u8], [u8]>(&self.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(
                                    &self.item_db,
                                    ref_item_id_bytes,
                                    updated_ref_item,
                                    &put::Flags::empty(),
                                )?;

                            // many-to-one
                            } else if !reader.as_vector().is_empty() {
                                bail!(
                                    "item cannot be deleted as children still hold back reference"
                                );
                            }
                        }
                    } else {
                        if field.indexed == Some(true) {
                            let mut key = BytesMut::new();
                            key.put_u8(model_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(&self.index_db, key.as_ref())?;
                        }

                        if field.queryable == Some(true) {
                            let mut key = BytesMut::new();
                            key.put_u8(model_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);
                                }
                                _ => {
                                    bail!("cannot query on {:?}", field.kind);
                                }
                            }

                            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]>(&self.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(
                                &self.query_db,
                                key.as_ref(),
                                updated_list,
                                &put::Flags::empty(),
                            )?;
                        }
                    }
                }

                if let Some(searchable) = self.model_search_indexes.get(model_idx as usize)
                    && let Some((schema, index, _reader)) = searchable
                {
                    let id_field = schema.get_field("0")?;
                    let del_term = Term::from_field_bytes(id_field, item_key.as_ref());

                    // todo: see if this is reusable so it doesn't have to be allocated on every write
                    let mut index_writer: IndexWriter = index.writer(15_000_000)?;
                    index_writer.delete_term(del_term);

                    index_writer.commit()?;
                    index_writer.wait_merging_threads()?;
                }

                access.del_key(&self.item_db, item_key.as_ref())?;
                access.del_key(&self.index_db, index_key.as_ref())?;
            }

            txn.commit()?;
        }

        Ok(())
    }
}
