1use super::{EthMappingsStore, SettingsStore, SettingsStoreExt};
5use crate::blocks::TipsetKey;
6use crate::db::PersistentStore;
7use crate::libp2p_bitswap::{BitswapStoreRead, BitswapStoreReadWrite};
8use crate::rpc::eth::types::EthHash;
9use crate::utils::db::car_stream::CarBlock;
10use crate::utils::multihash::prelude::*;
11use ahash::HashMap;
12use anyhow::Context as _;
13use cid::Cid;
14use fvm_ipld_blockstore::Blockstore;
15use itertools::Itertools;
16use parking_lot::RwLock;
17
18#[derive(Debug, Default)]
19pub struct MemoryDB {
20 blockchain_db: RwLock<HashMap<Cid, Vec<u8>>>,
21 blockchain_persistent_db: RwLock<HashMap<Cid, Vec<u8>>>,
22 settings_db: RwLock<HashMap<String, Vec<u8>>>,
23 pub eth_mappings_db: RwLock<HashMap<EthHash, Vec<u8>>>,
24}
25
26impl MemoryDB {
27 pub async fn export_forest_car<W: tokio::io::AsyncWrite + Unpin>(
28 &self,
29 writer: &mut W,
30 ) -> anyhow::Result<()> {
31 let roots =
32 SettingsStoreExt::read_obj::<TipsetKey>(self, crate::db::setting_keys::HEAD_KEY)?
33 .context("chain head is not tracked and cannot be exported")?
34 .into_cids();
35 let blocks = {
36 let blockchain_db = self.blockchain_db.read();
37 let blockchain_persistent_db = self.blockchain_persistent_db.read();
38 blockchain_db
39 .iter()
40 .chain(blockchain_persistent_db.iter())
41 .sorted_by_key(|&(&cid, _)| cid)
43 .map(|(&cid, data)| {
44 anyhow::Ok(CarBlock {
45 cid,
46 data: data.clone(),
47 })
48 })
49 .collect_vec()
50 };
51 let frames =
52 crate::db::car::forest::Encoder::compress_stream_default(futures::stream::iter(blocks));
53 crate::db::car::forest::Encoder::write(writer, roots, frames).await
54 }
55}
56
57impl SettingsStore for MemoryDB {
58 fn read_bin(&self, key: &str) -> anyhow::Result<Option<Vec<u8>>> {
59 Ok(self.settings_db.read().get(key).cloned())
60 }
61
62 fn write_bin(&self, key: &str, value: &[u8]) -> anyhow::Result<()> {
63 self.settings_db
64 .write()
65 .insert(key.to_owned(), value.to_vec());
66 Ok(())
67 }
68
69 fn exists(&self, key: &str) -> anyhow::Result<bool> {
70 Ok(self.settings_db.read().contains_key(key))
71 }
72
73 fn setting_keys(&self) -> anyhow::Result<Vec<String>> {
74 Ok(self.settings_db.read().keys().cloned().collect_vec())
75 }
76}
77
78impl EthMappingsStore for MemoryDB {
79 fn read_bin(&self, key: &EthHash) -> anyhow::Result<Option<Vec<u8>>> {
80 Ok(self.eth_mappings_db.read().get(key).cloned())
81 }
82
83 fn write_bin(&self, key: &EthHash, value: &[u8]) -> anyhow::Result<()> {
84 self.eth_mappings_db
85 .write()
86 .insert(key.to_owned(), value.to_vec());
87 Ok(())
88 }
89
90 fn exists(&self, key: &EthHash) -> anyhow::Result<bool> {
91 Ok(self.eth_mappings_db.read().contains_key(key))
92 }
93
94 fn get_message_cids(&self) -> anyhow::Result<Vec<(Cid, u64)>> {
95 let cids = self
96 .eth_mappings_db
97 .read()
98 .iter()
99 .filter_map(|(_, value)| fvm_ipld_encoding::from_slice::<(Cid, u64)>(value).ok())
100 .collect();
101
102 Ok(cids)
103 }
104
105 fn delete(&self, keys: Vec<EthHash>) -> anyhow::Result<()> {
106 let mut lock = self.eth_mappings_db.write();
107 for hash in keys.iter() {
108 lock.remove(hash);
109 }
110 Ok(())
111 }
112}
113
114impl Blockstore for MemoryDB {
115 fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
116 Ok(self.blockchain_db.read().get(k).cloned().or(self
117 .blockchain_persistent_db
118 .read()
119 .get(k)
120 .cloned()))
121 }
122
123 fn put_keyed(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
124 self.blockchain_db.write().insert(*k, block.to_vec());
125 Ok(())
126 }
127}
128
129impl PersistentStore for MemoryDB {
130 fn put_keyed_persistent(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
131 self.blockchain_persistent_db
132 .write()
133 .insert(*k, block.to_vec());
134 Ok(())
135 }
136}
137
138impl BitswapStoreRead for MemoryDB {
139 fn contains(&self, cid: &Cid) -> anyhow::Result<bool> {
140 Ok(self.blockchain_db.read().contains_key(cid))
141 }
142
143 fn get(&self, cid: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
144 Blockstore::get(self, cid)
145 }
146}
147
148impl BitswapStoreReadWrite for MemoryDB {
149 type Hashes = MultihashCode;
150
151 fn insert(&self, block: &crate::libp2p_bitswap::Block64<Self::Hashes>) -> anyhow::Result<()> {
152 self.put_keyed(block.cid(), block.data())
153 }
154}
155
156impl super::HeaviestTipsetKeyProvider for MemoryDB {
157 fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey> {
158 SettingsStoreExt::read_obj::<TipsetKey>(self, crate::db::setting_keys::HEAD_KEY)?
159 .context("head key not found")
160 }
161
162 fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()> {
163 SettingsStoreExt::write_obj(self, crate::db::setting_keys::HEAD_KEY, tsk)
164 }
165}
166
167#[cfg(test)]
168mod tests {
169 use super::*;
170 use crate::db::{car::ForestCar, setting_keys::HEAD_KEY};
171 use fvm_ipld_encoding::DAG_CBOR;
172 use multihash_codetable::Code::Blake2b256;
173 use nunny::vec as nonempty;
174
175 #[tokio::test]
176 async fn test_export_forest_car() {
177 let db = MemoryDB::default();
178 let record1 = b"non-persistent";
179 let key1 = Cid::new_v1(DAG_CBOR, Blake2b256.digest(record1.as_slice()));
180 db.put_keyed(&key1, record1.as_slice()).unwrap();
181
182 let record2 = b"persistent";
183 let key2 = Cid::new_v1(DAG_CBOR, Blake2b256.digest(record2.as_slice()));
184 db.put_keyed_persistent(&key2, record2.as_slice()).unwrap();
185
186 let mut car_db_bytes = vec![];
187 assert!(
188 db.export_forest_car(&mut car_db_bytes)
189 .await
190 .unwrap_err()
191 .to_string()
192 .contains("chain head is not tracked and cannot be exported")
193 );
194
195 db.write_obj(HEAD_KEY, &TipsetKey::from(nonempty![key1]))
196 .unwrap();
197
198 car_db_bytes.clear();
199 db.export_forest_car(&mut car_db_bytes).await.unwrap();
200
201 let car = ForestCar::new(car_db_bytes).unwrap();
202 assert_eq!(car.head_tipset_key(), &nonempty![key1]);
203 assert!(car.has(&key1).unwrap());
204 assert!(car.has(&key2).unwrap());
205 }
206}