use keri_core::database::LogDatabase;
use wasm_bindgen::prelude::*;
use wasm_bindgen_futures::spawn_local;
use super::IndexedDbError;
use std::cell::{Cell, RefCell};
use std::rc::Rc;
use std::sync::Arc;
use std::collections::HashMap;
use keri_core::event_message::signature::{Nontransferable, Transferable};
use keri_core::event::KeyEvent;
use keri_core::event_message::msg::KeriEvent;
use keri_core::event_message::signed_event_message::{ SignedEventMessage, SignedNontransferableReceipt };
use keri_core::database::timestamped::TimestampedSignedEventMessage;
use keri_core::prefix::IndexedSignature;
use said::SelfAddressingIdentifier;
const DB_NAME: &str = "log_db";
pub struct IndexedDbLogDatabase {
db_ref: Rc<RefCell<Option<web_sys::IdbDatabase>>>,
events: Rc<RefCell<HashMap<String, TimestampedSignedEventMessage>>>,
signatures: Rc<RefCell<HashMap<String, Vec<IndexedSignature>>>>,
trans_receipts: Rc<RefCell<HashMap<String, Vec<Transferable>>>>,
nontrans_receipts: Rc<RefCell<HashMap<String, Vec<Nontransferable>>>>,
pending_operations: Rc<RefCell<Vec<PendingOperation>>>,
flush_in_progress: Rc<Cell<bool>>,
}
unsafe impl Send for IndexedDbLogDatabase {}
unsafe impl Sync for IndexedDbLogDatabase {}
enum PendingOperation {
StoreEvent {
said: String,
event: Box<TimestampedSignedEventMessage>,
signatures: Vec<IndexedSignature>,
},
StoreNontransReceipt { said: String, receipts: Vec<Nontransferable> },
StoreTransReceipt { said: String, receipts: Vec<Transferable> },
RemoveReceipt {
said: String,
remaining_receipts: Vec<Nontransferable>,
},
}
impl IndexedDbLogDatabase {
fn init_db(&mut self) {
let window = web_sys::window().expect("should have a window");
if let Ok(Some(factory)) = window.indexed_db() {
let db_ref = self.db_ref.clone();
let events = self.events.clone();
let signatures = self.signatures.clone();
let trans_receipts = self.trans_receipts.clone();
let nontrans_receipts = self.nontrans_receipts.clone();
let pending_operations = self.pending_operations.clone();
let flush_in_progress = self.flush_in_progress.clone();
if let Ok(request) = factory.open(DB_NAME) {
let upgrade_needed_cb = Closure::wrap(Box::new(move |event: web_sys::IdbVersionChangeEvent| {
if let Some(db) = event.target()
.and_then(|t| t.dyn_into::<web_sys::IdbOpenDbRequest>().ok())
.and_then(|r| r.result().ok())
.and_then(|r| r.dyn_into::<web_sys::IdbDatabase>().ok())
{
let _ = db.create_object_store("events");
let _ = db.create_object_store("trans_receipts");
let _ = db.create_object_store("nontrans_receipts");
}
}) as Box<dyn FnMut(_)>);
request.set_onupgradeneeded(Some(upgrade_needed_cb.as_ref().unchecked_ref()));
upgrade_needed_cb.forget();
let success_cb = Closure::wrap(Box::new(move |event: web_sys::Event| {
if let Some(db) = event.target()
.and_then(|t| t.dyn_into::<web_sys::IdbOpenDbRequest>().ok())
.and_then(|r| r.result().ok())
.and_then(|r| r.dyn_into::<web_sys::IdbDatabase>().ok())
{
*db_ref.borrow_mut() = Some(db.clone());
load_events(&db, events.clone());
load_signatures(&db, signatures.clone());
load_trans_receipts(&db, trans_receipts.clone());
load_nontrans_receipts(&db, nontrans_receipts.clone());
setup_background_flush(db, pending_operations.clone(), flush_in_progress.clone());
}
}) as Box<dyn FnMut(_)>);
request.set_onsuccess(Some(success_cb.as_ref().unchecked_ref()));
success_cb.forget();
let error_cb = Closure::wrap(Box::new(|event: web_sys::Event| {
log::error!("Failed to open IndexedDB Log Database: {:?}", event);
}) as Box<dyn FnMut(_)>);
request.set_onerror(Some(error_cb.as_ref().unchecked_ref()));
error_cb.forget();
}
}
}
}
impl LogDatabase<'_> for IndexedDbLogDatabase {
type DatabaseType = ();
type Error = IndexedDbError;
type TransactionType = ();
fn new(_db: Arc<Self::DatabaseType>) -> Result<Self, IndexedDbError> {
let db_ref = Rc::new(RefCell::new(None));
let events = Rc::new(RefCell::new(HashMap::new()));
let signatures = Rc::new(RefCell::new(HashMap::new()));
let trans_receipts = Rc::new(RefCell::new(HashMap::new()));
let nontrans_receipts = Rc::new(RefCell::new(HashMap::new()));
let pending_operations = Rc::new(RefCell::new(Vec::new()));
let flush_in_progress = Rc::new(Cell::new(false));
let mut db = Self {
db_ref,
events,
signatures,
trans_receipts,
nontrans_receipts,
pending_operations,
flush_in_progress,
};
db.init_db();
Ok(db)
}
fn log_event(
&self,
_txn: &Self::TransactionType,
signed_event: &SignedEventMessage,
) -> Result<(), Self::Error> {
self.log_event_with_new_transaction(signed_event)
}
fn log_event_with_new_transaction(
&self,
signed_event: &SignedEventMessage,
) -> Result<(), Self::Error> {
let digest = signed_event
.event_message
.digest()
.map_err(|_| IndexedDbError::MissingDigest)?;
let said_str = digest.to_string();
let timestamped_event = TimestampedSignedEventMessage::new(signed_event.clone());
self.events
.borrow_mut()
.insert(said_str.clone(), timestamped_event.clone());
self.signatures
.borrow_mut()
.insert(said_str.clone(), signed_event.signatures.clone());
if let Some(receipts) = &signed_event.witness_receipts {
self.nontrans_receipts
.borrow_mut()
.insert(said_str.clone(), receipts.clone());
}
let mut pending = self.pending_operations.borrow_mut();
pending.push(PendingOperation::StoreEvent {
said: said_str.clone(),
event: Box::new(timestamped_event),
signatures: signed_event.signatures.clone(),
});
Ok(())
}
fn log_receipt(
&self,
_txn: &Self::TransactionType,
signed_receipt: &SignedNontransferableReceipt,
) -> Result<(), Self::Error> {
self.log_receipt_with_new_transaction(signed_receipt)
}
fn log_receipt_with_new_transaction(
&self,
signed_receipt: &SignedNontransferableReceipt,
) -> Result<(), Self::Error> {
let digest = &signed_receipt.body.receipted_event_digest;
let said_str = digest.to_string();
let mut receipts_map = self.nontrans_receipts
.borrow_mut();
let receipts = receipts_map.entry(said_str.clone()).or_default();
let mut new_receipts = Vec::new();
for receipt in &signed_receipt.signatures {
if !receipts.contains(receipt) {
receipts.push(receipt.clone());
new_receipts.push(receipt.clone());
}
}
if !new_receipts.is_empty() {
let mut pending = self.pending_operations.borrow_mut();
pending.push(PendingOperation::StoreNontransReceipt {
said: said_str.clone(),
receipts: new_receipts,
});
}
Ok(())
}
fn get_signed_event(
&self,
said: &SelfAddressingIdentifier,
) -> Result<Option<TimestampedSignedEventMessage>, Self::Error> {
let events = self.events.borrow();
let r = events.get(&said.to_string()).cloned();
Ok(r)
}
fn get_event(
&self,
said: &SelfAddressingIdentifier,
) -> Result<Option<KeriEvent<KeyEvent>>, Self::Error> {
let events = self.events.borrow();
match events.get(&said.to_string()) {
Some(signed_event) => Ok(Some(signed_event.signed_event_message.event_message.clone())),
None => Ok(None),
}
}
fn get_signatures(
&self,
said: &SelfAddressingIdentifier,
) -> Result<Option<impl Iterator<Item = IndexedSignature>>, Self::Error> {
let signatures = self.signatures.borrow();
match signatures.get(&said.to_string()) {
Some(sigs) => Ok(Some(sigs.clone().into_iter())),
None => Ok(None),
}
}
fn get_nontrans_couplets(
&self,
said: &SelfAddressingIdentifier,
) -> Result<Option<impl Iterator<Item = Nontransferable>>, Self::Error> {
let receipts = self.nontrans_receipts.borrow();
match receipts.get(&said.to_string()) {
Some(receipts) => Ok(Some(receipts.clone().into_iter())),
None => Ok(None),
}
}
fn get_trans_receipts(
&self,
said: &SelfAddressingIdentifier,
) -> Result<impl DoubleEndedIterator<Item = Transferable>, Self::Error> {
let receipts = self.trans_receipts.borrow();
match receipts.get(&said.to_string()) {
Some(receipts) => Ok(receipts.clone().into_iter()),
None => Ok(Vec::new().into_iter()),
}
}
fn remove_nontrans_receipt(
&self,
_txn_mode: &Self::TransactionType,
said: &SelfAddressingIdentifier,
nontrans: impl IntoIterator<Item = Nontransferable>,
) -> Result<(), Self::Error> {
let said_str = said.to_string();
let mut receipts_map = self.nontrans_receipts.borrow_mut();
if let Some(receipts) = receipts_map.get_mut(&said_str) {
let receipts_to_remove: Vec<Nontransferable> = nontrans.into_iter().collect();
receipts.retain(|r| !receipts_to_remove.contains(r));
let mut pending = self.pending_operations.borrow_mut();
pending.push(PendingOperation::RemoveReceipt {
said: said_str,
remaining_receipts: receipts.clone(),
});
}
Ok(())
}
fn remove_nontrans_receipt_with_new_transaction(
&self,
said: &SelfAddressingIdentifier,
nontrans: impl IntoIterator<Item = Nontransferable>,
) -> Result<(), Self::Error> {
self.remove_nontrans_receipt(&(), said, nontrans)
}
}
impl IndexedDbLogDatabase {
pub fn insert_trans_receipt(
&self,
said: &SelfAddressingIdentifier,
receipts: &[Transferable]
) -> Result<(), IndexedDbError> {
let said_str = said.to_string();
let mut receipts_map = self.trans_receipts.borrow_mut();
let existing_receipts = receipts_map.entry(said_str.clone()).or_default();
let mut new_receipts = Vec::new();
for receipt in receipts {
if !existing_receipts.contains(receipt) {
existing_receipts.push(receipt.clone());
new_receipts.push(receipt.clone());
}
}
if !new_receipts.is_empty() {
let mut pending = self.pending_operations.borrow_mut();
pending.push(PendingOperation::StoreTransReceipt {
said: said_str,
receipts: new_receipts,
});
}
Ok(())
}
pub fn insert_nontrans_receipt(
&self,
said: &SelfAddressingIdentifier,
receipts: &[Nontransferable]
) -> Result<(), IndexedDbError> {
let said_str = said.to_string();
let mut receipts_map = self.nontrans_receipts.borrow_mut();
let existing_receipts = receipts_map.entry(said_str.clone()).or_default();
let mut new_receipts = Vec::new();
for receipt in receipts {
if !existing_receipts.contains(receipt) {
existing_receipts.push(receipt.clone());
new_receipts.push(receipt.clone());
}
}
if !new_receipts.is_empty() {
let mut pending = self.pending_operations.borrow_mut();
pending.push(PendingOperation::StoreNontransReceipt {
said: said_str.clone(),
receipts: new_receipts,
});
}
Ok(())
}
pub fn get_nontrans_couplets_by_key(
&self,
key_prefix: &SelfAddressingIdentifier
) -> Result<Option<impl Iterator<Item = Nontransferable>>, IndexedDbError> {
let key_str = key_prefix.to_string();
let receipts_map = self.nontrans_receipts.borrow();
if let Some(receipts) = receipts_map.get(&key_str) {
if receipts.is_empty() {
Ok(None)
} else {
Ok(Some(receipts.clone().into_iter()))
}
} else {
Ok(None)
}
}
}
fn flush_pending_operations(
db: &web_sys::IdbDatabase,
operations: &Vec<PendingOperation>,
) {
for op in operations {
match op {
PendingOperation::StoreEvent { said, event, signatures } => {
if let Ok(transaction) = db.transaction_with_str_and_mode(
"events",
web_sys::IdbTransactionMode::Readwrite,
) {
if let Ok(store) = transaction.object_store("events") {
let value = js_sys::Object::new();
let _ = js_sys::Reflect::set(&value, &"said".into(), &said.clone().into());
let _ = js_sys::Reflect::set(
&value,
&"event".into(),
&JsValue::from_str(&format!("{:?}", serde_cbor::to_vec(&event).unwrap())),
);
if let Err(e) = store.put_with_key(&value, &said.into()) {
log::error!("Failed to store event in IndexedDB: {:?}", e);
}
}
}
if let Ok(transaction) = db.transaction_with_str_and_mode(
"signatures",
web_sys::IdbTransactionMode::Readwrite,
) {
if let Ok(store) = transaction.object_store("signatures") {
let value = js_sys::Object::new();
let _ = js_sys::Reflect::set(&value, &"said".into(), &said.into());
let _ = js_sys::Reflect::set(
&value,
&"signatures".into(),
&JsValue::from_str(&serde_json::to_string(&signatures).unwrap_or_default()),
);
if let Err(e) = store.put_with_key(&value, &said.into()) {
log::error!("Failed to store signatures in IndexedDB: {:?}", e);
}
}
}
},
PendingOperation::StoreNontransReceipt { said, receipts } => {
if let Ok(transaction) = db.transaction_with_str_and_mode(
"nontrans_receipts",
web_sys::IdbTransactionMode::Readwrite,
) {
if let Ok(store) = transaction.object_store("nontrans_receipts") {
let value = js_sys::Object::new();
let _ = js_sys::Reflect::set(&value, &"said".into(), &said.into());
let _ = js_sys::Reflect::set(
&value,
&"receipts".into(),
&JsValue::from_str(&serde_json::to_string(&receipts).unwrap_or_default()),
);
if let Err(e) = store.put_with_key(&value, &said.into()) {
log::error!("Failed to store receipts in IndexedDB: {:?}", e);
}
}
}
},
PendingOperation::StoreTransReceipt { said, receipts } => {
if let Ok(transaction) = db.transaction_with_str_and_mode(
"trans_receipts",
web_sys::IdbTransactionMode::Readwrite,
) {
if let Ok(store) = transaction.object_store("trans_receipts") {
let value = js_sys::Object::new();
let _ = js_sys::Reflect::set(&value, &"said".into(), &said.into());
let _ = js_sys::Reflect::set(
&value,
&"receipts".into(),
&JsValue::from_str(&serde_json::to_string(&receipts).unwrap_or_default()),
);
if let Err(e) = store.put_with_key(&value, &said.into()) {
log::error!("Failed to store transferable receipts in IndexedDB: {:?}", e);
}
}
}
},
PendingOperation::RemoveReceipt { said, remaining_receipts } => {
if let Ok(transaction) = db.transaction_with_str_and_mode(
"nontrans_receipts",
web_sys::IdbTransactionMode::Readwrite,
) {
if let Ok(store) = transaction.object_store("nontrans_receipts") {
if remaining_receipts.is_empty() {
if let Err(e) = store.delete(&said.into()) {
log::error!("Failed to delete receipts from IndexedDB: {:?}", e);
}
} else {
let value = js_sys::Object::new();
let _ = js_sys::Reflect::set(&value, &"said".into(), &said.into());
let _ = js_sys::Reflect::set(
&value,
&"receipts".into(),
&JsValue::from_str(&serde_json::to_string(&remaining_receipts).unwrap_or_default()),
);
if let Err(e) = store.put_with_key(&value, &said.into()) {
log::error!("Failed to update receipts in IndexedDB: {:?}", e);
}
}
}
}
}
}
}
}
fn load_events(db: &web_sys::IdbDatabase, events: Rc<RefCell<HashMap<String, TimestampedSignedEventMessage>>>) {
if let Ok(transaction) = db.transaction_with_str_and_mode(
"events",
web_sys::IdbTransactionMode::Readwrite,
) {
if let Ok(store) = transaction.object_store("events") {
if let Ok(request) = store.get_all() {
let callback = Closure::wrap(Box::new(move |event: web_sys::Event| {
if let Some(result) = event.target()
.and_then(|t| t.dyn_into::<web_sys::IdbRequest>().ok())
.and_then(|r| r.result().ok())
{
if let Ok(array) = result.dyn_into::<js_sys::Array>() {
let mut events_map = events.borrow_mut();
for i in 0..array.length() {
if let Ok(item) = array.get(i).dyn_into::<js_sys::Object>() {
if let (Some(said), Some(event_str)) = (
js_sys::Reflect::get(&item, &"said".into())
.ok()
.and_then(|v| v.as_string()),
js_sys::Reflect::get(&item, &"event".into())
.ok()
.and_then(|v| v.as_string()),
) {
let mut ev_vec: Vec<u8> = vec![];
let mut ev_str = event_str.clone();
ev_str.remove(0);
ev_str.pop();
for e in ev_str.split(", ") {
ev_vec.push(e.parse().unwrap());
}
if let Ok(event) = serde_cbor::from_slice::<TimestampedSignedEventMessage>(&ev_vec) {
events_map.insert(said, event);
} else {
log::warn!("Failed to parse event: {}", event_str);
}
}
}
}
}
}
}) as Box<dyn FnMut(_)>);
request.set_onsuccess(Some(callback.as_ref().unchecked_ref()));
callback.forget();
}
}
}
}
fn load_signatures(db: &web_sys::IdbDatabase, signatures: Rc<RefCell<HashMap<String, Vec<IndexedSignature>>>>) {
if let Ok(transaction) = db.transaction_with_str_and_mode(
"signatures",
web_sys::IdbTransactionMode::Readwrite,
) {
if let Ok(store) = transaction.object_store("signatures") {
if let Ok(request) = store.get_all() {
let callback = Closure::wrap(Box::new(move |event: web_sys::Event| {
if let Some(result) = event.target()
.and_then(|t| t.dyn_into::<web_sys::IdbRequest>().ok())
.and_then(|r| r.result().ok())
{
if let Ok(array) = result.dyn_into::<js_sys::Array>() {
let mut sigs_map = signatures.borrow_mut();
for i in 0..array.length() {
if let Ok(item) = array.get(i).dyn_into::<js_sys::Object>() {
if let (Ok(digest), Ok(sigs_json)) = (
js_sys::Reflect::get(&item, &"digest".into()),
js_sys::Reflect::get(&item, &"signatures".into())
) {
if let (Some(digest_str), Some(sigs_str)) = (
digest.as_string(),
sigs_json.as_string()
) {
if let Ok(sig_data) = serde_json::from_str::<Vec<IndexedSignature>>(&sigs_str) {
sigs_map.insert(digest_str, sig_data);
}
}
}
}
}
}
}
}) as Box<dyn FnMut(_)>);
request.set_onsuccess(Some(callback.as_ref().unchecked_ref()));
callback.forget();
}
}
}
}
fn load_trans_receipts(db: &web_sys::IdbDatabase, receipts: Rc<RefCell<std::collections::HashMap<String, Vec<Transferable>>>>) {
if let Ok(transaction) = db.transaction_with_str_and_mode(
"trans_receipts",
web_sys::IdbTransactionMode::Readwrite,
) {
if let Ok(store) = transaction.object_store("trans_receipts") {
if let Ok(request) = store.get_all() {
let callback = Closure::wrap(Box::new(move |event: web_sys::Event| {
if let Some(result) = event.target()
.and_then(|t| t.dyn_into::<web_sys::IdbRequest>().ok())
.and_then(|r| r.result().ok())
{
if let Ok(array) = result.dyn_into::<js_sys::Array>() {
let mut receipts_map = receipts.borrow_mut();
for i in 0..array.length() {
if let Ok(item) = array.get(i).dyn_into::<js_sys::Object>() {
if let (Ok(digest), Ok(receipts_json)) = (
js_sys::Reflect::get(&item, &"digest".into()),
js_sys::Reflect::get(&item, &"receipts".into())
) {
if let (Some(digest_str), Some(receipts_str)) = (
digest.as_string(),
receipts_json.as_string()
) {
if let Ok(trans_receipts) = serde_json::from_str::<Vec<Transferable>>(&receipts_str) {
receipts_map.insert(digest_str, trans_receipts);
}
}
}
}
}
}
}
}) as Box<dyn FnMut(_)>);
request.set_onsuccess(Some(callback.as_ref().unchecked_ref()));
callback.forget();
}
}
}
}
fn load_nontrans_receipts(db: &web_sys::IdbDatabase, receipts: Rc<RefCell<HashMap<String, Vec<Nontransferable>>>>) {
if let Ok(transaction) = db.transaction_with_str_and_mode(
"nontrans_receipts",
web_sys::IdbTransactionMode::Readwrite,
) {
if let Ok(store) = transaction.object_store("nontrans_receipts") {
if let Ok(request) = store.get_all() {
let callback = Closure::wrap(Box::new(move |event: web_sys::Event| {
if let Some(result) = event.target()
.and_then(|t| t.dyn_into::<web_sys::IdbRequest>().ok())
.and_then(|r| r.result().ok())
{
if let Ok(array) = result.dyn_into::<js_sys::Array>() {
let mut receipts_map = receipts.borrow_mut();
for i in 0..array.length() {
if let Ok(item) = array.get(i).dyn_into::<js_sys::Object>() {
if let (Ok(digest), Ok(receipts_json)) = (
js_sys::Reflect::get(&item, &"digest".into()),
js_sys::Reflect::get(&item, &"receipts".into())
) {
if let (Some(digest_str), Some(receipts_str)) = (
digest.as_string(),
receipts_json.as_string()
) {
if let Ok(nontrans_receipts) = serde_json::from_str::<Vec<Nontransferable>>(&receipts_str) {
receipts_map.insert(digest_str, nontrans_receipts);
}
}
}
}
}
}
}
}) as Box<dyn FnMut(_)>);
request.set_onsuccess(Some(callback.as_ref().unchecked_ref()));
callback.forget();
}
}
}
}
fn setup_background_flush(
db: web_sys::IdbDatabase,
pending_ops: Rc<RefCell<Vec<PendingOperation>>>,
flush_flag: Rc<Cell<bool>>
) {
let interval_callback = Closure::wrap(Box::new(move || {
if !flush_flag.get() {
flush_flag.set(true);
spawn_local({
let db = db.clone();
let pending = pending_ops.clone();
let flag = flush_flag.clone();
async move {
let ops_to_flush = {
let mut pending_borrow = pending.borrow_mut();
if pending_borrow.is_empty() {
Vec::new()
} else {
pending_borrow.drain(..).collect::<Vec<_>>()
}
};
if !ops_to_flush.is_empty() {
flush_pending_operations(&db, &ops_to_flush);
}
flag.set(false);
}
});
}
}) as Box<dyn FnMut()>);
let window = web_sys::window().unwrap();
let _ = window.set_interval_with_callback_and_timeout_and_arguments_0(
interval_callback.as_ref().unchecked_ref(),
1000
);
interval_callback.forget();
}