Split low-level LMDB code out from Store (big refactor)

chorus-lib/src/store/lmdb/mod.rs

@@ -0,0 +1,710 @@
+use crate::config::Config;
+use crate::error::Error;
+use crate::ip::{HashedIp, IpData};
+use crate::types::{Event, Id, Kind, Pubkey, Time};
+use heed::types::{OwnedType, UnalignedSlice, Unit, U8};
+use heed::{Database, Env, EnvFlags, EnvOpenOptions, RoIter, RoRange, RoTxn, RwTxn};
+use speedy::{Readable, Writable};
+use std::fs;
+use std::ops::Bound;
+
+mod retired;
+
+#[derive(Debug)]
+pub struct Lmdb {
+    env: Env,
+    general: Database<UnalignedSlice<u8>, UnalignedSlice<u8>>,
+    i_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
+    ci_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
+    tc_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
+    ac_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
+    akc_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
+    atc_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
+    ktc_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
+
+    // this is for events deleted by other events
+    deleted_ids: Database<UnalignedSlice<u8>, Unit>,
+    approved_events: Database<UnalignedSlice<u8>, U8>,
+    approved_pubkeys: Database<UnalignedSlice<u8>, U8>,
+    ip_data: Database<UnalignedSlice<u8>, UnalignedSlice<u8>>,
+}
+
+impl Lmdb {
+    pub fn new(config: &Config) -> Result<Lmdb, Error> {
+        let mut builder = EnvOpenOptions::new();
+        unsafe {
+            builder.flags(EnvFlags::NO_TLS);
+        }
+        builder.max_dbs(32);
+        builder.map_size(1048576 * 1024 * 24); // 24 GB
+
+        let dir = format!("{}/lmdb", &config.data_directory);
+        fs::create_dir_all(&dir)?;
+
+        let env = match builder.open(&dir) {
+            Ok(env) => env,
+            Err(e) => {
+                log::error!("Unable to open LMDB at {}", dir);
+                return Err(e.into());
+            }
+        };
+
+        // Open/Create maps
+        let mut txn = env.write_txn()?;
+        let general = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, UnalignedSlice<u8>>()
+            .create(&mut txn)?;
+        let i_index = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
+            .name("ids")
+            .create(&mut txn)?;
+        let ci_index = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
+            .name("ci")
+            .create(&mut txn)?;
+        let tc_index = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
+            .name("tci")
+            .create(&mut txn)?;
+        let ac_index = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
+            .name("aci")
+            .create(&mut txn)?;
+        let akc_index = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
+            .name("akci")
+            .create(&mut txn)?;
+        let atc_index = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
+            .name("atci")
+            .create(&mut txn)?;
+        let ktc_index = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
+            .name("ktci")
+            .create(&mut txn)?;
+        let deleted_ids = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, Unit>()
+            .name("deleted-ids")
+            .create(&mut txn)?;
+        let approved_events = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, U8>()
+            .name("approved-events")
+            .create(&mut txn)?;
+        let approved_pubkeys = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, U8>()
+            .name("approved-pubkeys")
+            .create(&mut txn)?;
+        let ip_data = env
+            .database_options()
+            .types::<UnalignedSlice<u8>, UnalignedSlice<u8>>()
+            .name("ip_data")
+            .create(&mut txn)?;
+        txn.commit()?;
+
+        let lmdb = Lmdb {
+            env,
+            general,
+            i_index,
+            ci_index,
+            tc_index,
+            ac_index,
+            akc_index,
+            atc_index,
+            ktc_index,
+            deleted_ids,
+            approved_events,
+            approved_pubkeys,
+            ip_data,
+        };
+
+        Ok(lmdb)
+    }
+
+    /// Sync the data to disk. This happens periodically, but sometimes it's useful to force
+    /// it.
+    pub fn sync(&self) -> Result<(), Error> {
+        self.env.force_sync()?;
+        Ok(())
+    }
+
+    /// Get a read transaction
+    pub fn read_txn(&self) -> Result<RoTxn, Error> {
+        Ok(self.env.read_txn()?)
+    }
+
+    /// Get a write transaction
+    pub fn write_txn(&self) -> Result<RwTxn, Error> {
+        Ok(self.env.write_txn()?)
+    }
+
+    pub fn log_stats(&self, txn: &RoTxn) {
+        if let Ok(count) = self.i_index.len(txn) {
+            log::info!("Index: id ({} entries, {} bytes)", count, count * (32 + 8));
+        }
+        if let Ok(count) = self.ci_index.len(txn) {
+            log::info!(
+                "Index: created_at+id ({} entries, {} bytes)",
+                count,
+                count * (40 + 8)
+            );
+        }
+
+        if let Ok(count) = self.tc_index.len(txn) {
+            log::info!(
+                "Index: tag+created_at+id ({} entries, {} bytes)",
+                count,
+                count * (223 + 8)
+            );
+        }
+        if let Ok(count) = self.ac_index.len(txn) {
+            log::info!(
+                "Index: author+created_at+id ({} entries, {} bytes)",
+                count,
+                count * (72 + 8)
+            );
+        }
+        if let Ok(count) = self.akc_index.len(txn) {
+            log::info!(
+                "Index: author+kind+created_at+id ({} entries, {} bytes)",
+                count,
+                count * (74 + 8)
+            );
+        }
+        if let Ok(count) = self.atc_index.len(txn) {
+            log::info!(
+                "Index: author+tags+created_at+id ({} entries, {} bytes)",
+                count,
+                count * (255 + 8)
+            );
+        }
+        if let Ok(count) = self.ktc_index.len(txn) {
+            log::info!(
+                "Index: kind+tags+created_at+id ({} entries, {} bytes)",
+                count,
+                count * (225 + 8)
+            );
+        }
+
+        if let Ok(count) = self.deleted_ids.len(txn) {
+            log::info!("{} deleted events", count);
+        }
+        if let Ok(count) = self.ip_data.len(txn) {
+            log::info!("{count} IP addresses reputationally tracked");
+        }
+    }
+
+    pub fn get_migration_level(&self, txn: &RoTxn<'_>) -> Result<u32, Error> {
+        let zero_bytes = 0_u32.to_be_bytes();
+        let migration_level_bytes = self
+            .general
+            .get(txn, b"migration_level")?
+            .unwrap_or(zero_bytes.as_slice());
+        Ok(u32::from_be_bytes(
+            migration_level_bytes[..4].try_into().unwrap(),
+        ))
+    }
+
+    pub fn set_migration_level(&self, txn: &mut RwTxn<'_>, level: u32) -> Result<(), Error> {
+        self.general
+            .put(txn, b"migration_level", level.to_be_bytes().as_slice())?;
+
+        Ok(())
+    }
+
+    // Index the event
+    pub fn index(&self, txn: &mut RwTxn<'_>, event: &Event, offset: usize) -> Result<(), Error> {
+        // Index by id
+        self.i_index.put(txn, event.id().0.as_slice(), &offset)?;
+
+        // Index by created_at and id
+        self.ci_index.put(
+            txn,
+            &Self::key_ci_index(event.created_at(), event.id()),
+            &offset,
+        )?;
+
+        // Index by author and kind (with created_at and id)
+        self.akc_index.put(
+            txn,
+            &Self::key_akc_index(event.pubkey(), event.kind(), event.created_at(), event.id()),
+            &offset,
+        )?;
+
+        self.ac_index.put(
+            txn,
+            &Self::key_ac_index(event.pubkey(), event.created_at(), event.id()),
+            &offset,
+        )?;
+
+        for mut tsi in event.tags()?.iter() {
+            if let Some(tagname) = tsi.next() {
+                // FIXME make sure it is a letter too
+                if tagname.len() == 1 {
+                    if let Some(tagvalue) = tsi.next() {
+                        // Index by tag (with created_at and id)
+                        self.tc_index.put(
+                            txn,
+                            &Self::key_tc_index(
+                                tagname[0],
+                                tagvalue,
+                                event.created_at(),
+                                event.id(),
+                            ),
+                            &offset,
+                        )?;
+
+                        // Index by author and tag (with created_at and id)
+                        self.atc_index.put(
+                            txn,
+                            &Self::key_atc_index(
+                                event.pubkey(),
+                                tagname[0],
+                                tagvalue,
+                                event.created_at(),
+                                event.id(),
+                            ),
+                            &offset,
+                        )?;
+
+                        // Index by kind and tag (with created_at and id)
+                        self.ktc_index.put(
+                            txn,
+                            &Self::key_ktc_index(
+                                event.kind(),
+                                tagname[0],
+                                tagvalue,
+                                event.created_at(),
+                                event.id(),
+                            ),
+                            &offset,
+                        )?;
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    // Remove the event from all indexes (except the 'id' index)
+    pub fn deindex(&self, txn: &mut RwTxn<'_>, event: &Event) -> Result<(), Error> {
+        for mut tsi in event.tags()?.iter() {
+            if let Some(tagname) = tsi.next() {
+                // FIXME make sure it is a letter too
+                if tagname.len() == 1 {
+                    if let Some(tagvalue) = tsi.next() {
+                        // Index by author and tag (with created_at and id)
+                        self.atc_index.delete(
+                            txn,
+                            &Self::key_atc_index(
+                                event.pubkey(),
+                                tagname[0],
+                                tagvalue,
+                                event.created_at(),
+                                event.id(),
+                            ),
+                        )?;
+
+                        // Index by kind and tag (with created_at and id)
+                        self.ktc_index.delete(
+                            txn,
+                            &Self::key_ktc_index(
+                                event.kind(),
+                                tagname[0],
+                                tagvalue,
+                                event.created_at(),
+                                event.id(),
+                            ),
+                        )?;
+
+                        // Index by tag (with created_at and id)
+                        self.tc_index.delete(
+                            txn,
+                            &Self::key_tc_index(
+                                tagname[0],
+                                tagvalue,
+                                event.created_at(),
+                                event.id(),
+                            ),
+                        )?;
+                    }
+                }
+            }
+        }
+
+        self.ac_index.delete(
+            txn,
+            &Self::key_ac_index(event.pubkey(), event.created_at(), event.id()),
+        )?;
+
+        self.ci_index
+            .delete(txn, &Self::key_ci_index(event.created_at(), event.id()))?;
+
+        self.akc_index.delete(
+            txn,
+            &Self::key_akc_index(event.pubkey(), event.kind(), event.created_at(), event.id()),
+        )?;
+
+        // We leave it in the id map. If someone wants to load the replaced event by id
+        // they can still do it.
+        // self.i_index.delete(&mut txn, event.id().0.as_slice())?;
+
+        Ok(())
+    }
+
+    pub fn deindex_id(&self, txn: &mut RwTxn<'_>, id: Id) -> Result<(), Error> {
+        self.i_index.delete(txn, id.0.as_slice())?;
+        Ok(())
+    }
+
+    pub fn get_offset_by_id(&self, txn: &RoTxn<'_>, id: Id) -> Result<Option<usize>, Error> {
+        Ok(self.i_index.get(txn, id.0.as_slice())?)
+    }
+
+    pub fn is_deleted(&self, txn: &RoTxn<'_>, id: Id) -> Result<bool, Error> {
+        Ok(self.deleted_ids.get(txn, id.as_slice())?.is_some())
+    }
+
+    pub fn mark_deleted(&self, txn: &mut RwTxn<'_>, id: Id) -> Result<(), Error> {
+        self.deleted_ids.put(txn, id.as_slice(), &())?;
+        Ok(())
+    }
+
+    pub fn get_ip_data(&self, ip: HashedIp) -> Result<IpData, Error> {
+        let key = &ip.0;
+        let txn = self.read_txn()?;
+        let bytes = match self.ip_data.get(&txn, key)? {
+            Some(b) => b,
+            None => return Ok(Default::default()),
+        };
+        Ok(IpData::read_from_buffer(bytes)?)
+    }
+
+    pub fn update_ip_data(&self, ip: HashedIp, data: &IpData) -> Result<(), Error> {
+        let key = &ip.0;
+        let mut txn = self.write_txn()?;
+        let bytes = data.write_to_vec()?;
+        self.ip_data.put(&mut txn, key, &bytes)?;
+        txn.commit()?;
+        Ok(())
+    }
+
+    pub fn mark_event_approval(&self, id: Id, approval: bool) -> Result<(), Error> {
+        let mut txn = self.write_txn()?;
+        self.approved_events
+            .put(&mut txn, id.0.as_slice(), &(approval as u8))?;
+        txn.commit()?;
+        Ok(())
+    }
+
+    pub fn clear_event_approval(&self, id: Id) -> Result<(), Error> {
+        let mut txn = self.write_txn()?;
+        self.approved_events.delete(&mut txn, id.0.as_slice())?;
+        txn.commit()?;
+        Ok(())
+    }
+
+    pub fn get_event_approval(&self, id: Id) -> Result<Option<bool>, Error> {
+        let txn = self.read_txn()?;
+        Ok(self
+            .approved_events
+            .get(&txn, id.0.as_slice())?
+            .map(|u| u != 0))
+    }
+
+    pub fn dump_event_approvals(&self) -> Result<Vec<(Id, bool)>, Error> {
+        let mut output: Vec<(Id, bool)> = Vec::new();
+        let txn = self.read_txn()?;
+        for i in self.approved_events.iter(&txn)? {
+            let (key, val) = i?;
+            let id = Id(key.try_into().unwrap());
+            let approval: bool = val != 0;
+            output.push((id, approval));
+        }
+        Ok(output)
+    }
+
+    pub fn mark_pubkey_approval(&self, pubkey: Pubkey, approval: bool) -> Result<(), Error> {
+        let mut txn = self.write_txn()?;
+        self.approved_pubkeys
+            .put(&mut txn, pubkey.0.as_slice(), &(approval as u8))?;
+        txn.commit()?;
+        Ok(())
+    }
+
+    pub fn clear_pubkey_approval(&self, pubkey: Pubkey) -> Result<(), Error> {
+        let mut txn = self.write_txn()?;
+        self.approved_pubkeys
+            .delete(&mut txn, pubkey.0.as_slice())?;
+        txn.commit()?;
+        Ok(())
+    }
+
+    pub fn get_pubkey_approval(&self, pubkey: Pubkey) -> Result<Option<bool>, Error> {
+        let txn = self.read_txn()?;
+        Ok(self
+            .approved_pubkeys
+            .get(&txn, pubkey.0.as_slice())?
+            .map(|u| u != 0))
+    }
+
+    pub fn dump_pubkey_approvals(&self) -> Result<Vec<(Pubkey, bool)>, Error> {
+        let mut output: Vec<(Pubkey, bool)> = Vec::new();
+        let txn = self.read_txn()?;
+        for i in self.approved_pubkeys.iter(&txn)? {
+            let (key, val) = i?;
+            let pubkey = Pubkey(key.try_into().unwrap());
+            let approval: bool = val != 0;
+            output.push((pubkey, approval));
+        }
+        Ok(output)
+    }
+
+    pub fn i_iter<'a>(
+        &'a self,
+        txn: &'a RoTxn,
+    ) -> Result<RoIter<'_, UnalignedSlice<u8>, OwnedType<usize>>, Error> {
+        Ok(self.i_index.iter(txn)?)
+    }
+
+    pub fn ci_iter<'a>(
+        &'a self,
+        since: Time,
+        until: Time,
+        txn: &'a RoTxn,
+    ) -> Result<RoRange<'_, UnalignedSlice<u8>, OwnedType<usize>>, Error> {
+        let start_prefix = Self::key_ci_index(until, Id([0; 32]));
+        let end_prefix = Self::key_ci_index(since, Id([255; 32]));
+        let range = (
+            Bound::Included(&*start_prefix),
+            Bound::Excluded(&*end_prefix),
+        );
+        Ok(self.ci_index.range(txn, &range)?)
+    }
+
+    pub fn tc_iter<'a>(
+        &'a self,
+        tagbyte: u8,
+        tagvalue: &[u8],
+        since: Time,
+        until: Time,
+        txn: &'a RoTxn,
+    ) -> Result<RoRange<'_, UnalignedSlice<u8>, OwnedType<usize>>, Error> {
+        let start_prefix = Self::key_tc_index(
+            tagbyte,
+            tagvalue,
+            until, // scan goes backwards in time
+            Id([0; 32]),
+        );
+        let end_prefix = Self::key_tc_index(tagbyte, tagvalue, since, Id([255; 32]));
+        let range = (
+            Bound::Included(&*start_prefix),
+            Bound::Excluded(&*end_prefix),
+        );
+        Ok(self.tc_index.range(txn, &range)?)
+    }
+
+    pub fn ac_iter<'a>(
+        &'a self,
+        author: Pubkey,
+        since: Time,
+        until: Time,
+        txn: &'a RoTxn,
+    ) -> Result<RoRange<'_, UnalignedSlice<u8>, OwnedType<usize>>, Error> {
+        let start_prefix = Self::key_ac_index(author, until, Id([0; 32]));
+        let end_prefix = Self::key_ac_index(author, since, Id([255; 32]));
+        let range = (
+            Bound::Included(&*start_prefix),
+            Bound::Excluded(&*end_prefix),
+        );
+        Ok(self.ac_index.range(txn, &range)?)
+    }
+
+    pub fn akc_iter<'a>(
+        &'a self,
+        author: Pubkey,
+        kind: Kind,
+        since: Time,
+        until: Time,
+        txn: &'a RoTxn,
+    ) -> Result<RoRange<'_, UnalignedSlice<u8>, OwnedType<usize>>, Error> {
+        let start_prefix = Self::key_akc_index(author, kind, until, Id([0; 32]));
+        let end_prefix = Self::key_akc_index(author, kind, since, Id([255; 32]));
+        let range = (
+            Bound::Included(&*start_prefix),
+            Bound::Excluded(&*end_prefix),
+        );
+        Ok(self.akc_index.range(txn, &range)?)
+    }
+
+    pub fn atc_iter<'a>(
+        &'a self,
+        author: Pubkey,
+        tagbyte: u8,
+        tagvalue: &[u8],
+        since: Time,
+        until: Time,
+        txn: &'a RoTxn,
+    ) -> Result<RoRange<'_, UnalignedSlice<u8>, OwnedType<usize>>, Error> {
+        let start_prefix = Self::key_atc_index(
+            author,
+            tagbyte,
+            tagvalue,
+            until, // scan goes backwards in time
+            Id([0; 32]),
+        );
+        let end_prefix = Self::key_atc_index(author, tagbyte, tagvalue, since, Id([255; 32]));
+        let range = (
+            Bound::Included(&*start_prefix),
+            Bound::Excluded(&*end_prefix),
+        );
+        Ok(self.atc_index.range(txn, &range)?)
+    }
+
+    pub fn ktc_iter<'a>(
+        &'a self,
+        kind: Kind,
+        tagbyte: u8,
+        tagvalue: &[u8],
+        since: Time,
+        until: Time,
+        txn: &'a RoTxn,
+    ) -> Result<RoRange<'_, UnalignedSlice<u8>, OwnedType<usize>>, Error> {
+        let start_prefix = Self::key_ktc_index(
+            kind,
+            tagbyte,
+            tagvalue,
+            until, // scan goes backwards in time
+            Id([0; 32]),
+        );
+        let end_prefix = Self::key_ktc_index(kind, tagbyte, tagvalue, since, Id([255; 32]));
+        let range = (
+            Bound::Included(&*start_prefix),
+            Bound::Excluded(&*end_prefix),
+        );
+        Ok(self.ktc_index.range(txn, &range)?)
+    }
+
+    fn key_ci_index(created_at: Time, id: Id) -> Vec<u8> {
+        let mut key: Vec<u8> =
+            Vec::with_capacity(std::mem::size_of::<Time>() + std::mem::size_of::<Id>());
+        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
+        key.extend(id.as_slice());
+        key
+    }
+
+    // For looking up event by Tag
+    // tagletter(1) + fixlentag(182) + reversecreatedat(8) + id(32)
+    fn key_tc_index(letter: u8, tag_value: &[u8], created_at: Time, id: Id) -> Vec<u8> {
+        const PADLEN: usize = 182;
+        let mut key: Vec<u8> =
+            Vec::with_capacity(PADLEN + std::mem::size_of::<Time>() + std::mem::size_of::<Id>());
+        key.push(letter);
+        if tag_value.len() <= PADLEN {
+            key.extend(tag_value);
+            key.extend(core::iter::repeat(0).take(PADLEN - tag_value.len()));
+        } else {
+            key.extend(&tag_value[..PADLEN]);
+        }
+        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
+        key.extend(id.as_slice());
+        key
+    }
+
+    // For looking up event by Author
+    // author(32) + reversecreatedat(8) + id(32)
+    fn key_ac_index(author: Pubkey, created_at: Time, id: Id) -> Vec<u8> {
+        let mut key: Vec<u8> = Vec::with_capacity(
+            std::mem::size_of::<Pubkey>() + std::mem::size_of::<Time>() + std::mem::size_of::<Id>(),
+        );
+        key.extend(author.as_slice());
+        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
+        key.extend(id.as_slice());
+        key
+    }
+
+    // For looking up event by Author and Kind
+    // author(32) + kind(2) + reversecreatedat(8) + id(32)
+    fn key_akc_index(author: Pubkey, kind: Kind, created_at: Time, id: Id) -> Vec<u8> {
+        let mut key: Vec<u8> = Vec::with_capacity(
+            std::mem::size_of::<Pubkey>()
+                + std::mem::size_of::<Kind>()
+                + std::mem::size_of::<Time>()
+                + std::mem::size_of::<Id>(),
+        );
+        key.extend(author.as_slice());
+        key.extend(kind.0.to_be_bytes());
+        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
+        key.extend(id.as_slice());
+        key
+    }
+
+    // For looking up event by Author and Tag
+    // author(32) + tagletter(1) + fixlentag(182) + reversecreatedat(8) + id(32)
+    fn key_atc_index(
+        author: Pubkey,
+        letter: u8,
+        tag_value: &[u8],
+        created_at: Time,
+        id: Id,
+    ) -> Vec<u8> {
+        const PADLEN: usize = 182;
+        let mut key: Vec<u8> = Vec::with_capacity(
+            std::mem::size_of::<Pubkey>()
+                + PADLEN
+                + std::mem::size_of::<Time>()
+                + std::mem::size_of::<Id>(),
+        );
+        key.extend(author.as_slice());
+        key.push(letter);
+        if tag_value.len() <= PADLEN {
+            key.extend(tag_value);
+            key.extend(core::iter::repeat(0).take(PADLEN - tag_value.len()));
+        } else {
+            key.extend(&tag_value[..PADLEN]);
+        }
+        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
+        key.extend(id.as_slice());
+        key
+    }
+
+    // For looking up event by Kind and Tag
+    // kind(2) + tagletter(1) + fixlentag(182) + reversecreatedat(8) + id(32)
+    fn key_ktc_index(
+        kind: Kind,
+        letter: u8,
+        tag_value: &[u8],
+        created_at: Time,
+        id: Id,
+    ) -> Vec<u8> {
+        const PADLEN: usize = 182;
+        let mut key: Vec<u8> = Vec::with_capacity(
+            std::mem::size_of::<Kind>()
+                + PADLEN
+                + std::mem::size_of::<Time>()
+                + std::mem::size_of::<Id>(),
+        );
+        key.extend(kind.0.to_be_bytes());
+        key.push(letter);
+        if tag_value.len() <= PADLEN {
+            key.extend(tag_value);
+            key.extend(core::iter::repeat(0).take(PADLEN - tag_value.len()));
+        } else {
+            key.extend(&tag_value[..PADLEN]);
+        }
+        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
+        key.extend(id.as_slice());
+        key
+    }
+}

chorus-lib/src/store/lmdb/retired.rs

@@ -0,0 +1,88 @@
+use super::Lmdb;
+use crate::error::Error;
+use crate::types::Event;
+use heed::byteorder::BigEndian;
+use heed::types::{UnalignedSlice, Unit, U64};
+use heed::{Database, RwTxn};
+
+impl Lmdb {
+    pub fn deleted_offsets(
+        &self,
+        txn: &mut RwTxn,
+    ) -> Result<Database<U64<BigEndian>, Unit>, Error> {
+        Ok(self
+            .env
+            .database_options()
+            .types::<U64<BigEndian>, Unit>()
+            .name("deleted_offsets")
+            .create(txn)?)
+    }
+
+    pub fn deleted_events(
+        &self,
+        txn: &mut RwTxn,
+    ) -> Result<Database<UnalignedSlice<u8>, Unit>, Error> {
+        Ok(self
+            .env
+            .database_options()
+            .types::<UnalignedSlice<u8>, Unit>()
+            .name("deleted-events")
+            .create(txn)?)
+    }
+
+    // used in migrate_to_1
+    pub fn index_ci_only(
+        &self,
+        txn: &mut RwTxn<'_>,
+        event: &Event,
+        offset: usize,
+    ) -> Result<(), Error> {
+        self.ci_index.put(
+            txn,
+            &Self::key_ci_index(event.created_at(), event.id()),
+            &offset,
+        )?;
+
+        Ok(())
+    }
+
+    // used in migrate_to_2
+    pub fn index_ac_only(
+        &self,
+        txn: &mut RwTxn<'_>,
+        event: &Event,
+        offset: usize,
+    ) -> Result<(), Error> {
+        self.ac_index.put(
+            txn,
+            &Self::key_ac_index(event.pubkey(), event.created_at(), event.id()),
+            &offset,
+        )?;
+
+        Ok(())
+    }
+
+    // used in migrate_to_2
+    pub fn index_tc_only(
+        &self,
+        txn: &mut RwTxn<'_>,
+        tagbyte: u8,
+        tagvalue: &[u8],
+        event: &Event,
+        offset: usize,
+    ) -> Result<(), Error> {
+        self.tc_index.put(
+            txn,
+            &Self::key_tc_index(tagbyte, tagvalue, event.created_at(), event.id()),
+            &offset,
+        )?;
+
+        Ok(())
+    }
+
+    // used in migrate_to_3
+    pub fn clear_ip_data(&self, txn: &mut RwTxn<'_>) -> Result<(), Error> {
+        self.ip_data.clear(txn)?;
+        Ok(())
+    }
+}

chorus-lib/src/store/migrations.rs

@@ -1,35 +1,23 @@
 use super::Store;
 use crate::error::Error;
 use crate::types::Id;
-use heed::byteorder::BigEndian;
-use heed::types::{UnalignedSlice, Unit, U64};
 use heed::RwTxn;
 
 pub const CURRENT_MIGRATION_LEVEL: u32 = 5;
 
 impl Store {
     pub fn migrate(&self) -> Result<(), Error> {
-        let mut txn = self.env.write_txn()?;
+        let mut txn = self.lmdb.write_txn()?;
 
-        let mut migration_level = {
-            let zero_bytes = 0_u32.to_be_bytes();
-            let migration_level_bytes = self
-                .general
-                .get(&txn, b"migration_level")?
-                .unwrap_or(zero_bytes.as_slice());
-            u32::from_be_bytes(migration_level_bytes[..4].try_into().unwrap())
-        };
+        let mut migration_level = self.lmdb.get_migration_level(&txn)?;
 
         log::info!("Storage migration level = {}", migration_level);
 
         while migration_level < CURRENT_MIGRATION_LEVEL {
             self.migrate_to(&mut txn, migration_level + 1)?;
             migration_level += 1;
-            self.general.put(
-                &mut txn,
-                b"migration_level",
-                migration_level.to_be_bytes().as_slice(),
-            )?;
+
+            self.lmdb.set_migration_level(&mut txn, migration_level)?
         }
 
         txn.commit()?;
@@ -53,16 +41,12 @@ impl Store {
 
     // Populate ci_index
     fn migrate_to_1(&self, txn: &mut RwTxn<'_>) -> Result<(), Error> {
-        let loop_txn = self.env.read_txn()?;
-        let iter = self.i_index.iter(&loop_txn)?;
+        let loop_txn = self.lmdb.read_txn()?;
+        let iter = self.lmdb.i_iter(&loop_txn)?;
         for result in iter {
             let (_key, offset) = result?;
             let event = self.events.get_event_by_offset(offset)?;
-            self.ci_index.put(
-                txn,
-                &Self::key_ci_index(event.created_at(), event.id()),
-                &offset,
-            )?;
+            self.lmdb.index_ci_only(txn, &event, offset)?;
         }
 
         Ok(())
@@ -70,34 +54,22 @@ impl Store {
 
     // Populate tc_index and ac_index
     fn migrate_to_2(&self, txn: &mut RwTxn<'_>) -> Result<(), Error> {
-        let loop_txn = self.env.read_txn()?;
-        let iter = self.i_index.iter(&loop_txn)?;
+        let loop_txn = self.lmdb.read_txn()?;
+        let iter = self.lmdb.i_iter(&loop_txn)?;
         for result in iter {
             let (_key, offset) = result?;
             let event = self.events.get_event_by_offset(offset)?;
 
             // Add to ac_index
-            self.ac_index.put(
-                txn,
-                &Self::key_ac_index(event.pubkey(), event.created_at(), event.id()),
-                &offset,
-            )?;
+            self.lmdb.index_ac_only(txn, &event, offset)?;
 
             // Add to tc_index
             for mut tsi in event.tags()?.iter() {
                 if let Some(tagname) = tsi.next() {
                     if tagname.len() == 1 {
                         if let Some(tagvalue) = tsi.next() {
-                            self.tc_index.put(
-                                txn,
-                                &Self::key_tc_index(
-                                    tagname[0],
-                                    tagvalue,
-                                    event.created_at(),
-                                    event.id(),
-                                ),
-                                &offset,
-                            )?;
+                            self.lmdb
+                                .index_tc_only(txn, tagname[0], tagvalue, &event, offset)?;
                         }
                     }
                 }
@@ -109,30 +81,20 @@ impl Store {
 
     // Clear IP data (we are hashing now)
     fn migrate_to_3(&self, txn: &mut RwTxn<'_>) -> Result<(), Error> {
-        self.ip_data.clear(txn)?;
+        self.lmdb.clear_ip_data(txn)?;
         Ok(())
     }
 
     // Clear deleted_offsets (now retired)
     fn migrate_to_4(&self, txn: &mut RwTxn<'_>) -> Result<(), Error> {
-        let deleted_offsets = self
-            .env
-            .database_options()
-            .types::<U64<BigEndian>, Unit>()
-            .name("deleted_offsets")
-            .create(txn)?;
+        let deleted_offsets = self.lmdb.deleted_offsets(txn)?;
         deleted_offsets.clear(txn)?;
         Ok(())
     }
 
     // Move data from deleted_events to deleted_ids
     fn migrate_to_5(&self, txn: &mut RwTxn<'_>) -> Result<(), Error> {
-        let deleted_events = self
-            .env
-            .database_options()
-            .types::<UnalignedSlice<u8>, Unit>()
-            .name("deleted-events")
-            .create(txn)?;
+        let deleted_events = self.lmdb.deleted_events(txn)?;
 
         let mut ids: Vec<Id> = Vec::new();
 
@@ -143,7 +105,7 @@ impl Store {
         }
 
         for id in ids.drain(..) {
-            self.deleted_ids.put(txn, id.as_slice(), &())?;
+            self.lmdb.mark_deleted(txn, id)?;
         }
 
         deleted_events.clear(txn)?;

chorus-lib/src/store/mod.rs

@@ -1,208 +1,51 @@
 pub mod event_store;
 pub use event_store::EventStore;
 
+mod lmdb;
+use lmdb::Lmdb;
+
 mod migrations;
 
 use crate::config::Config;
 use crate::error::{ChorusError, Error};
 use crate::ip::{HashedIp, IpData};
 use crate::types::{Event, Filter, Id, Kind, Pubkey, Time};
-use heed::types::{OwnedType, UnalignedSlice, Unit, U8};
-use heed::{Database, Env, EnvFlags, EnvOpenOptions, RwTxn};
-use speedy::{Readable, Writable};
+use heed::RwTxn;
 use std::collections::BTreeSet;
-use std::fs;
-use std::ops::Bound;
 
 #[derive(Debug)]
 pub struct Store {
-    general: Database<UnalignedSlice<u8>, UnalignedSlice<u8>>,
+    lmdb: Lmdb,
     events: EventStore,
-    env: Env,
-    i_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
-    ci_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
-    tc_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
-    ac_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
-    akc_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
-    atc_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
-    ktc_index: Database<UnalignedSlice<u8>, OwnedType<usize>>,
-
-    // this is for events deleted by other events
-    deleted_ids: Database<UnalignedSlice<u8>, Unit>,
-    approved_events: Database<UnalignedSlice<u8>, U8>,
-    approved_pubkeys: Database<UnalignedSlice<u8>, U8>,
-    ip_data: Database<UnalignedSlice<u8>, UnalignedSlice<u8>>,
 }
 
 impl Store {
     /// Setup persistent storage
     pub fn new(config: &Config) -> Result<Store, Error> {
-        let mut builder = EnvOpenOptions::new();
-        unsafe {
-            builder.flags(EnvFlags::NO_TLS);
-        }
-        builder.max_dbs(32);
-        builder.map_size(1048576 * 1024 * 24); // 24 GB
+        let lmdb = Lmdb::new(config)?;
 
-        let dir = format!("{}/lmdb", &config.data_directory);
-        fs::create_dir_all(&dir)?;
-
-        let env = match builder.open(&dir) {
-            Ok(env) => env,
-            Err(e) => {
-                log::error!("Unable to open LMDB at {}", dir);
-                return Err(e.into());
-            }
+        let events = {
+            let event_map_file = format!("{}/event.map", &config.data_directory);
+            EventStore::new(event_map_file)?
         };
 
-        // Open/Create maps
-        let mut txn = env.write_txn()?;
-        let general = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, UnalignedSlice<u8>>()
-            .create(&mut txn)?;
-        let i_index = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
-            .name("ids")
-            .create(&mut txn)?;
-        let ci_index = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
-            .name("ci")
-            .create(&mut txn)?;
-        let tc_index = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
-            .name("tci")
-            .create(&mut txn)?;
-        let ac_index = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
-            .name("aci")
-            .create(&mut txn)?;
-        let akc_index = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
-            .name("akci")
-            .create(&mut txn)?;
-        let atc_index = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
-            .name("atci")
-            .create(&mut txn)?;
-        let ktc_index = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, OwnedType<usize>>()
-            .name("ktci")
-            .create(&mut txn)?;
-        let deleted_ids = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, Unit>()
-            .name("deleted-ids")
-            .create(&mut txn)?;
-        let approved_events = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, U8>()
-            .name("approved-events")
-            .create(&mut txn)?;
-        let approved_pubkeys = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, U8>()
-            .name("approved-pubkeys")
-            .create(&mut txn)?;
-        let ip_data = env
-            .database_options()
-            .types::<UnalignedSlice<u8>, UnalignedSlice<u8>>()
-            .name("ip_data")
-            .create(&mut txn)?;
-
-        if let Ok(count) = i_index.len(&txn) {
-            log::info!("Index: id ({} entries, {} bytes)", count, count * (32 + 8));
-        }
-        if let Ok(count) = ci_index.len(&txn) {
-            log::info!(
-                "Index: created_at+id ({} entries, {} bytes)",
-                count,
-                count * (40 + 8)
-            );
-        }
-
-        if let Ok(count) = tc_index.len(&txn) {
-            log::info!(
-                "Index: tag+created_at+id ({} entries, {} bytes)",
-                count,
-                count * (223 + 8)
-            );
-        }
-        if let Ok(count) = ac_index.len(&txn) {
-            log::info!(
-                "Index: author+created_at+id ({} entries, {} bytes)",
-                count,
-                count * (72 + 8)
-            );
-        }
-        if let Ok(count) = akc_index.len(&txn) {
-            log::info!(
-                "Index: author+kind+created_at+id ({} entries, {} bytes)",
-                count,
-                count * (74 + 8)
-            );
-        }
-        if let Ok(count) = atc_index.len(&txn) {
-            log::info!(
-                "Index: author+tags+created_at+id ({} entries, {} bytes)",
-                count,
-                count * (255 + 8)
-            );
-        }
-        if let Ok(count) = ktc_index.len(&txn) {
-            log::info!(
-                "Index: kind+tags+created_at+id ({} entries, {} bytes)",
-                count,
-                count * (225 + 8)
-            );
-        }
-
-        if let Ok(count) = deleted_ids.len(&txn) {
-            log::info!("{} deleted events", count);
-        }
-        if let Ok(count) = ip_data.len(&txn) {
-            log::info!("{count} IP addresses reputationally tracked");
-        }
-
-        txn.commit()?;
-
-        let event_map_file = format!("{}/event.map", &config.data_directory);
-        let events = EventStore::new(event_map_file)?;
-
-        let store = Store {
-            general,
-            events,
-            env,
-            i_index,
-            ci_index,
-            tc_index,
-            ac_index,
-            akc_index,
-            atc_index,
-            ktc_index,
-            deleted_ids,
-            approved_events,
-            approved_pubkeys,
-            ip_data,
-        };
+        let store = Store { lmdb, events };
 
         // This is in migrations.rs
         store.migrate()?;
 
+        {
+            let txn = store.lmdb.read_txn()?;
+            store.lmdb.log_stats(&txn);
+        }
+
         Ok(store)
     }
 
     /// Sync the data to disk. This happens periodically, but sometimes it's useful to force
     /// it.
     pub fn sync(&self) -> Result<(), Error> {
-        self.env.force_sync()?;
+        self.lmdb.sync()?;
         Ok(())
     }
 
@@ -218,14 +61,14 @@ impl Store {
     pub fn store_event(&self, event: &Event) -> Result<usize, Error> {
         // TBD: should we validate the event?
 
-        let mut txn = self.env.write_txn()?;
+        let mut txn = self.lmdb.write_txn()?;
         let offset;
 
         // Only if it doesn't already exist
-        if self.i_index.get(&txn, event.id().0.as_slice())?.is_none() {
+        if self.lmdb.get_offset_by_id(&txn, event.id())?.is_none() {
             // Reject event if it was deleted
             {
-                if self.deleted_ids.get(&txn, event.id().as_slice())?.is_some() {
+                if self.lmdb.is_deleted(&txn, event.id())? {
                     return Err(ChorusError::Deleted.into());
                 }
             }
@@ -235,7 +78,7 @@ impl Store {
 
             // Index the event
             if !event.kind().is_ephemeral() {
-                self.index(&mut txn, event, offset)?;
+                self.lmdb.index(&mut txn, event, offset)?;
             }
 
             // If replaceable or parameterized replaceable,
@@ -264,7 +107,7 @@ impl Store {
                     if let Some(id_hex) = tag.next() {
                         if let Ok(id) = Id::read_hex(id_hex) {
                             // Add deletion pair to the event_deleted table
-                            self.deleted_ids.put(txn, id.as_slice(), &())?;
+                            self.lmdb.mark_deleted(txn, id)?;
 
                             // Delete pair
                             if let Some(target) = self.get_event_by_id(id)? {
@@ -288,8 +131,8 @@ impl Store {
 
     /// Get an event by Id
     pub fn get_event_by_id(&self, id: Id) -> Result<Option<Event>, Error> {
-        let txn = self.env.read_txn()?;
-        if let Some(offset) = self.i_index.get(&txn, id.0.as_slice())? {
+        let txn = self.lmdb.read_txn()?;
+        if let Some(offset) = self.lmdb.get_offset_by_id(&txn, id)? {
             Some(self.events.get_event_by_offset(offset)).transpose()
         } else {
             Ok(None)
@@ -306,7 +149,7 @@ impl Store {
     where
         F: Fn(&Event) -> bool,
     {
-        let txn = self.env.read_txn()?;
+        let txn = self.lmdb.read_txn()?;
 
         // We insert into a BTreeSet to keep them time-ordered
         let mut output: BTreeSet<Event> = BTreeSet::new();
@@ -332,20 +175,9 @@ impl Store {
 
             for author in filter.authors() {
                 for kind in filter.kinds() {
-                    let iter = {
-                        let start_prefix = Self::key_akc_index(
-                            author,
-                            kind,
-                            filter.until(), // scan goes backwards in time
-                            Id([0; 32]),
-                        );
-                        let end_prefix = Self::key_akc_index(author, kind, since, Id([255; 32]));
-                        let range = (
-                            Bound::Included(&*start_prefix),
-                            Bound::Excluded(&*end_prefix),
-                        );
-                        self.akc_index.range(&txn, &range)?
-                    };
+                    let iter = self
+                        .lmdb
+                        .akc_iter(author, kind, since, filter.until(), &txn)?;
 
                     // Count how many we have found of this author-kind pair, so we
                     // can possibly update `since`
@@ -399,27 +231,14 @@ impl Store {
                 for mut tag in tags.iter() {
                     if let Some(tag0) = tag.next() {
                         if let Some(tagvalue) = tag.next() {
-                            let iter = {
-                                let start_prefix = Self::key_atc_index(
-                                    author,
-                                    tag0[0],
-                                    tagvalue,
-                                    filter.until(), // scan goes backwards in time
-                                    Id([0; 32]),
-                                );
-                                let end_prefix = Self::key_atc_index(
-                                    author,
-                                    tag0[0],
-                                    tagvalue,
-                                    since,
-                                    Id([255; 32]),
-                                );
-                                let range = (
-                                    Bound::Included(&*start_prefix),
-                                    Bound::Excluded(&*end_prefix),
-                                );
-                                self.atc_index.range(&txn, &range)?
-                            };
+                            let iter = self.lmdb.atc_iter(
+                                author,
+                                tag0[0],
+                                tagvalue,
+                                since,
+                                filter.until(),
+                                &txn,
+                            )?;
 
                             // Count how many we have found of this author-tag pair, so we
                             // can possibly update `since`
@@ -466,27 +285,14 @@ impl Store {
                 for mut tag in tags.iter() {
                     if let Some(tag0) = tag.next() {
                         if let Some(tagvalue) = tag.next() {
-                            let iter = {
-                                let start_prefix = Self::key_ktc_index(
-                                    kind,
-                                    tag0[0],
-                                    tagvalue,
-                                    filter.until(), // scan goes backwards in time
-                                    Id([0; 32]),
-                                );
-                                let end_prefix = Self::key_ktc_index(
-                                    kind,
-                                    tag0[0],
-                                    tagvalue,
-                                    since,
-                                    Id([255; 32]),
-                                );
-                                let range = (
-                                    Bound::Included(&*start_prefix),
-                                    Bound::Excluded(&*end_prefix),
-                                );
-                                self.ktc_index.range(&txn, &range)?
-                            };
+                            let iter = self.lmdb.ktc_iter(
+                                kind,
+                                tag0[0],
+                                tagvalue,
+                                since,
+                                filter.until(),
+                                &txn,
+                            )?;
 
                             // Count how many we have found of this kind-tag pair, so we
                             // can possibly update `since`
@@ -532,21 +338,9 @@ impl Store {
             for mut tag in tags.iter() {
                 if let Some(tag0) = tag.next() {
                     if let Some(tagvalue) = tag.next() {
-                        let iter = {
-                            let start_prefix = Self::key_tc_index(
-                                tag0[0],
-                                tagvalue,
-                                filter.until(), // scan goes backwards in time
-                                Id([0; 32]),
-                            );
-                            let end_prefix =
-                                Self::key_tc_index(tag0[0], tagvalue, since, Id([255; 32]));
-                            let range = (
-                                Bound::Included(&*start_prefix),
-                                Bound::Excluded(&*end_prefix),
-                            );
-                            self.tc_index.range(&txn, &range)?
-                        };
+                        let iter =
+                            self.lmdb
+                                .tc_iter(tag0[0], tagvalue, since, filter.until(), &txn)?;
 
                         let mut rangecount = 0;
 
@@ -582,19 +376,7 @@ impl Store {
             let mut since = filter.since();
 
             for author in filter.authors() {
-                let iter = {
-                    let start_prefix = Self::key_ac_index(
-                        author,
-                        filter.until(), // scan goes backwards in time
-                        Id([0; 32]),
-                    );
-                    let end_prefix = Self::key_ac_index(author, since, Id([255; 32]));
-                    let range = (
-                        Bound::Included(&*start_prefix),
-                        Bound::Excluded(&*end_prefix),
-                    );
-                    self.ac_index.range(&txn, &range)?
-                };
+                let iter = self.lmdb.ac_iter(author, since, filter.until(), &txn)?;
 
                 let mut rangecount = 0;
 
@@ -635,17 +417,7 @@ impl Store {
 
             // This is INEFFICIENT as it scans through many events
 
-            let start_prefix = Self::key_ci_index(
-                filter.until(), // scan goes backwards
-                Id([0; 32]),
-            );
-            let end_prefix = Self::key_ci_index(filter.since(), Id([255; 32]));
-            let range = (
-                Bound::Included(&*start_prefix),
-                Bound::Excluded(&*end_prefix),
-            );
-
-            let iter = self.ci_index.range(&txn, &range)?;
+            let iter = self.lmdb.ci_iter(filter.since(), filter.until(), &txn)?;
             for result in iter {
                 if output.len() >= filter.limit() as usize {
                     break;
@@ -675,7 +447,7 @@ impl Store {
     /// This does not add to the deleted_ids record, which is for events
     /// that are deleted by other events
     fn delete_by_id(&self, txn: &mut RwTxn<'_>, id: Id) -> Result<(), Error> {
-        if let Some(offset) = self.i_index.get(txn, id.0.as_slice())? {
+        if let Some(offset) = self.lmdb.get_offset_by_id(txn, id)? {
             self.delete_by_offset(txn, offset)?;
         }
 
@@ -693,181 +465,34 @@ impl Store {
         let event = self.events.get_event_by_offset(offset)?;
 
         // Remove from indexes
-        self.deindex(txn, &event)?;
+        self.lmdb.deindex(txn, &event)?;
 
         // Also remove from the id index
-        self.i_index.delete(txn, event.id().0.as_slice())?;
+        self.lmdb.deindex_id(txn, event.id())?;
 
         Ok(())
     }
 
     // This deletes an event without marking it as having been deleted by another event
     pub fn delete_event(&self, id: Id) -> Result<(), Error> {
-        let mut txn = self.env.write_txn()?;
+        let mut txn = self.lmdb.write_txn()?;
         self.delete_by_id(&mut txn, id)?;
         txn.commit()?;
         Ok(())
     }
 
-    // Index the event
-    fn index(&self, txn: &mut RwTxn<'_>, event: &Event, offset: usize) -> Result<(), Error> {
-        // Index by id
-        self.i_index.put(txn, event.id().0.as_slice(), &offset)?;
-
-        // Index by created_at and id
-        self.ci_index.put(
-            txn,
-            &Self::key_ci_index(event.created_at(), event.id()),
-            &offset,
-        )?;
-
-        // Index by author and kind (with created_at and id)
-        self.akc_index.put(
-            txn,
-            &Self::key_akc_index(event.pubkey(), event.kind(), event.created_at(), event.id()),
-            &offset,
-        )?;
-
-        self.ac_index.put(
-            txn,
-            &Self::key_ac_index(event.pubkey(), event.created_at(), event.id()),
-            &offset,
-        )?;
-
-        for mut tsi in event.tags()?.iter() {
-            if let Some(tagname) = tsi.next() {
-                // FIXME make sure it is a letter too
-                if tagname.len() == 1 {
-                    if let Some(tagvalue) = tsi.next() {
-                        // Index by tag (with created_at and id)
-                        self.tc_index.put(
-                            txn,
-                            &Self::key_tc_index(
-                                tagname[0],
-                                tagvalue,
-                                event.created_at(),
-                                event.id(),
-                            ),
-                            &offset,
-                        )?;
-
-                        // Index by author and tag (with created_at and id)
-                        self.atc_index.put(
-                            txn,
-                            &Self::key_atc_index(
-                                event.pubkey(),
-                                tagname[0],
-                                tagvalue,
-                                event.created_at(),
-                                event.id(),
-                            ),
-                            &offset,
-                        )?;
-
-                        // Index by kind and tag (with created_at and id)
-                        self.ktc_index.put(
-                            txn,
-                            &Self::key_ktc_index(
-                                event.kind(),
-                                tagname[0],
-                                tagvalue,
-                                event.created_at(),
-                                event.id(),
-                            ),
-                            &offset,
-                        )?;
-                    }
-                }
-            }
-        }
-
-        Ok(())
-    }
-
-    // Remove the event from all indexes (except the 'id' index)
-    fn deindex(&self, txn: &mut RwTxn<'_>, event: &Event) -> Result<(), Error> {
-        for mut tsi in event.tags()?.iter() {
-            if let Some(tagname) = tsi.next() {
-                // FIXME make sure it is a letter too
-                if tagname.len() == 1 {
-                    if let Some(tagvalue) = tsi.next() {
-                        // Index by author and tag (with created_at and id)
-                        self.atc_index.delete(
-                            txn,
-                            &Self::key_atc_index(
-                                event.pubkey(),
-                                tagname[0],
-                                tagvalue,
-                                event.created_at(),
-                                event.id(),
-                            ),
-                        )?;
-
-                        // Index by kind and tag (with created_at and id)
-                        self.ktc_index.delete(
-                            txn,
-                            &Self::key_ktc_index(
-                                event.kind(),
-                                tagname[0],
-                                tagvalue,
-                                event.created_at(),
-                                event.id(),
-                            ),
-                        )?;
-
-                        // Index by tag (with created_at and id)
-                        self.tc_index.delete(
-                            txn,
-                            &Self::key_tc_index(
-                                tagname[0],
-                                tagvalue,
-                                event.created_at(),
-                                event.id(),
-                            ),
-                        )?;
-                    }
-                }
-            }
-        }
-
-        self.ac_index.delete(
-            txn,
-            &Self::key_ac_index(event.pubkey(), event.created_at(), event.id()),
-        )?;
-
-        self.ci_index
-            .delete(txn, &Self::key_ci_index(event.created_at(), event.id()))?;
-
-        self.akc_index.delete(
-            txn,
-            &Self::key_akc_index(event.pubkey(), event.kind(), event.created_at(), event.id()),
-        )?;
-
-        // We leave it in the id map. If someone wants to load the replaced event by id
-        // they can still do it.
-        // self.i_index.delete(&mut txn, event.id().0.as_slice())?;
-
-        Ok(())
-    }
-
     // If the event is replaceable or parameterized replaceable
     // this deletes all the events in that group except the most recent one.
     fn delete_replaced(&self, txn: &mut RwTxn<'_>, event: &Event) -> Result<(), Error> {
         if event.kind().is_replaceable() {
-            let start_prefix = Self::key_akc_index(
+            let loop_txn = self.lmdb.read_txn()?;
+            let iter = self.lmdb.akc_iter(
                 event.pubkey(),
                 event.kind(),
-                Time::max(), // database is ordered in reverse time
-                Id([0; 32]),
-            );
-            let end_prefix =
-                Self::key_akc_index(event.pubkey(), event.kind(), Time::min(), Id([255; 32]));
-            let range = (
-                Bound::Included(&*start_prefix),
-                Bound::Excluded(&*end_prefix),
-            );
-            let loop_txn = self.env.read_txn()?;
-            let iter = self.akc_index.range(&loop_txn, &range)?;
+                Time::min(),
+                Time::max(),
+                &loop_txn,
+            )?;
             let mut first = true;
             for result in iter {
                 // Keep the first result
@@ -884,21 +509,15 @@ impl Store {
         } else if event.kind().is_parameterized_replaceable() {
             let tags = event.tags()?;
             if let Some(identifier) = tags.get_value(b"d") {
-                let start_prefix =
-                    Self::key_atc_index(event.pubkey(), b'd', identifier, Time::max(), Id([0; 32]));
-                let end_prefix = Self::key_atc_index(
+                let loop_txn = self.lmdb.read_txn()?;
+                let iter = self.lmdb.atc_iter(
                     event.pubkey(),
                     b'd',
                     identifier,
                     Time::min(),
-                    Id([255; 32]),
-                );
-                let range = (
-                    Bound::Included(&*start_prefix),
-                    Bound::Excluded(&*end_prefix),
-                );
-                let loop_txn = self.env.read_txn()?;
-                let iter = self.atc_index.range(&loop_txn, &range)?;
+                    Time::max(),
+                    &loop_txn,
+                )?;
                 let mut first = true;
                 for result in iter {
                     // Keep the first result
@@ -919,204 +538,42 @@ impl Store {
     }
 
     pub fn get_ip_data(&self, ip: HashedIp) -> Result<IpData, Error> {
-        let key = &ip.0;
-        let txn = self.env.read_txn()?;
-        let bytes = match self.ip_data.get(&txn, key)? {
-            Some(b) => b,
-            None => return Ok(Default::default()),
-        };
-        Ok(IpData::read_from_buffer(bytes)?)
+        self.lmdb.get_ip_data(ip)
     }
 
     pub fn update_ip_data(&self, ip: HashedIp, data: &IpData) -> Result<(), Error> {
-        let key = &ip.0;
-        let mut txn = self.env.write_txn()?;
-        let bytes = data.write_to_vec()?;
-        self.ip_data.put(&mut txn, key, &bytes)?;
-        txn.commit()?;
-        Ok(())
+        self.lmdb.update_ip_data(ip, data)
     }
 
     pub fn mark_event_approval(&self, id: Id, approval: bool) -> Result<(), Error> {
-        let mut txn = self.env.write_txn()?;
-        self.approved_events
-            .put(&mut txn, id.0.as_slice(), &(approval as u8))?;
-        txn.commit()?;
-        Ok(())
+        self.lmdb.mark_event_approval(id, approval)
     }
 
     pub fn clear_event_approval(&self, id: Id) -> Result<(), Error> {
-        let mut txn = self.env.write_txn()?;
-        self.approved_events.delete(&mut txn, id.0.as_slice())?;
-        txn.commit()?;
-        Ok(())
+        self.lmdb.clear_event_approval(id)
     }
 
     pub fn get_event_approval(&self, id: Id) -> Result<Option<bool>, Error> {
-        let txn = self.env.read_txn()?;
-        Ok(self
-            .approved_events
-            .get(&txn, id.0.as_slice())?
-            .map(|u| u != 0))
+        self.lmdb.get_event_approval(id)
     }
 
     pub fn dump_event_approvals(&self) -> Result<Vec<(Id, bool)>, Error> {
-        let mut output: Vec<(Id, bool)> = Vec::new();
-        let txn = self.env.read_txn()?;
-        for i in self.approved_events.iter(&txn)? {
-            let (key, val) = i?;
-            let id = Id(key.try_into().unwrap());
-            let approval: bool = val != 0;
-            output.push((id, approval));
-        }
-        Ok(output)
+        self.lmdb.dump_event_approvals()
     }
 
     pub fn mark_pubkey_approval(&self, pubkey: Pubkey, approval: bool) -> Result<(), Error> {
-        let mut txn = self.env.write_txn()?;
-        self.approved_pubkeys
-            .put(&mut txn, pubkey.0.as_slice(), &(approval as u8))?;
-        txn.commit()?;
-        Ok(())
+        self.lmdb.mark_pubkey_approval(pubkey, approval)
     }
 
     pub fn clear_pubkey_approval(&self, pubkey: Pubkey) -> Result<(), Error> {
-        let mut txn = self.env.write_txn()?;
-        self.approved_pubkeys
-            .delete(&mut txn, pubkey.0.as_slice())?;
-        txn.commit()?;
-        Ok(())
+        self.lmdb.clear_pubkey_approval(pubkey)
     }
 
     pub fn get_pubkey_approval(&self, pubkey: Pubkey) -> Result<Option<bool>, Error> {
-        let txn = self.env.read_txn()?;
-        Ok(self
-            .approved_pubkeys
-            .get(&txn, pubkey.0.as_slice())?
-            .map(|u| u != 0))
+        self.lmdb.get_pubkey_approval(pubkey)
     }
 
     pub fn dump_pubkey_approvals(&self) -> Result<Vec<(Pubkey, bool)>, Error> {
-        let mut output: Vec<(Pubkey, bool)> = Vec::new();
-        let txn = self.env.read_txn()?;
-        for i in self.approved_pubkeys.iter(&txn)? {
-            let (key, val) = i?;
-            let pubkey = Pubkey(key.try_into().unwrap());
-            let approval: bool = val != 0;
-            output.push((pubkey, approval));
-        }
-        Ok(output)
-    }
-
-    fn key_ci_index(created_at: Time, id: Id) -> Vec<u8> {
-        let mut key: Vec<u8> =
-            Vec::with_capacity(std::mem::size_of::<Time>() + std::mem::size_of::<Id>());
-        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
-        key.extend(id.as_slice());
-        key
-    }
-
-    // For looking up event by Tag
-    // tagletter(1) + fixlentag(182) + reversecreatedat(8) + id(32)
-    fn key_tc_index(letter: u8, tag_value: &[u8], created_at: Time, id: Id) -> Vec<u8> {
-        const PADLEN: usize = 182;
-        let mut key: Vec<u8> =
-            Vec::with_capacity(PADLEN + std::mem::size_of::<Time>() + std::mem::size_of::<Id>());
-        key.push(letter);
-        if tag_value.len() <= PADLEN {
-            key.extend(tag_value);
-            key.extend(core::iter::repeat(0).take(PADLEN - tag_value.len()));
-        } else {
-            key.extend(&tag_value[..PADLEN]);
-        }
-        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
-        key.extend(id.as_slice());
-        key
-    }
-
-    // For looking up event by Author
-    // author(32) + reversecreatedat(8) + id(32)
-    fn key_ac_index(author: Pubkey, created_at: Time, id: Id) -> Vec<u8> {
-        let mut key: Vec<u8> = Vec::with_capacity(
-            std::mem::size_of::<Pubkey>() + std::mem::size_of::<Time>() + std::mem::size_of::<Id>(),
-        );
-        key.extend(author.as_slice());
-        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
-        key.extend(id.as_slice());
-        key
-    }
-
-    // For looking up event by Author and Kind
-    // author(32) + kind(2) + reversecreatedat(8) + id(32)
-    fn key_akc_index(author: Pubkey, kind: Kind, created_at: Time, id: Id) -> Vec<u8> {
-        let mut key: Vec<u8> = Vec::with_capacity(
-            std::mem::size_of::<Pubkey>()
-                + std::mem::size_of::<Kind>()
-                + std::mem::size_of::<Time>()
-                + std::mem::size_of::<Id>(),
-        );
-        key.extend(author.as_slice());
-        key.extend(kind.0.to_be_bytes());
-        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
-        key.extend(id.as_slice());
-        key
-    }
-
-    // For looking up event by Author and Tag
-    // author(32) + tagletter(1) + fixlentag(182) + reversecreatedat(8) + id(32)
-    fn key_atc_index(
-        author: Pubkey,
-        letter: u8,
-        tag_value: &[u8],
-        created_at: Time,
-        id: Id,
-    ) -> Vec<u8> {
-        const PADLEN: usize = 182;
-        let mut key: Vec<u8> = Vec::with_capacity(
-            std::mem::size_of::<Pubkey>()
-                + PADLEN
-                + std::mem::size_of::<Time>()
-                + std::mem::size_of::<Id>(),
-        );
-        key.extend(author.as_slice());
-        key.push(letter);
-        if tag_value.len() <= PADLEN {
-            key.extend(tag_value);
-            key.extend(core::iter::repeat(0).take(PADLEN - tag_value.len()));
-        } else {
-            key.extend(&tag_value[..PADLEN]);
-        }
-        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
-        key.extend(id.as_slice());
-        key
-    }
-
-    // For looking up event by Kind and Tag
-    // kind(2) + tagletter(1) + fixlentag(182) + reversecreatedat(8) + id(32)
-    fn key_ktc_index(
-        kind: Kind,
-        letter: u8,
-        tag_value: &[u8],
-        created_at: Time,
-        id: Id,
-    ) -> Vec<u8> {
-        const PADLEN: usize = 182;
-        let mut key: Vec<u8> = Vec::with_capacity(
-            std::mem::size_of::<Kind>()
-                + PADLEN
-                + std::mem::size_of::<Time>()
-                + std::mem::size_of::<Id>(),
-        );
-        key.extend(kind.0.to_be_bytes());
-        key.push(letter);
-        if tag_value.len() <= PADLEN {
-            key.extend(tag_value);
-            key.extend(core::iter::repeat(0).take(PADLEN - tag_value.len()));
-        } else {
-            key.extend(&tag_value[..PADLEN]);
-        }
-        key.extend((u64::MAX - created_at.0).to_be_bytes().as_slice());
-        key.extend(id.as_slice());
-        key
+        self.lmdb.dump_pubkey_approvals()
     }
 }