+18

ufos/src/storage.rs

Reviewed

··· 1 1 + // use crate::store_types::CountsValue; 1 2 use crate::{error::StorageError, Cursor, EventBatch, UFOsRecord}; 2 3 use jetstream::exports::{Did, Nsid}; 3 4 use std::path::Path; 4 5 5 6 pub type StorageResult<T> = Result<T, StorageError>; 6 7 8 8 + // #[derive(Debug)] 9 9 + // pub enum RollupTask { 10 10 + // CollectionRollup { 11 11 + // live_counts_key_bytes: Vec<u8>, 12 12 + // counts: CountsValue, 13 13 + // }, 14 14 + // DeleteAccount(Did), 15 15 + // } 16 16 + 7 17 pub trait StorageWhatever<R: StoreReader, W: StoreWriter, C> { 8 18 // TODO: extract this 9 19 fn init( ··· 21 31 &mut self, 22 32 event_batch: EventBatch<LIMIT>, 23 33 ) -> StorageResult<()>; 34 34 + 24 35 fn trim_collection(&mut self, collection: &Nsid, limit: usize) -> StorageResult<()>; 36 36 + 25 37 fn delete_account(&mut self, did: &Did) -> StorageResult<usize>; 38 38 + 39 39 + // fn rollup_tasks( 40 40 + // &mut self, 41 41 + // from_cursor: Cursor, 42 42 + // ) -> impl Iterator<Item = StorageResult<(RollupTask, Cursor)>>; 26 43 } 27 44 28 45 pub trait StoreReader: Clone { 29 46 fn get_counts_by_collection(&self, collection: &Nsid) -> StorageResult<(u64, u64)>; 47 47 + 30 48 fn get_records_by_collections( 31 49 &self, 32 50 collections: &[&Nsid],

+57 -1

ufos/src/storage_fjall.rs

Reviewed

··· 1157 1157 #[cfg(test)] 1158 1158 mod tests { 1159 1159 use super::*; 1160 1160 - use crate::UFOsCommit; 1160 1160 + use crate::{DeleteAccount, UFOsCommit}; 1161 1161 use jetstream::events::{CommitEvent, CommitOp}; 1162 1162 use jetstream::exports::Cid; 1163 1163 use serde_json::value::RawValue; ··· 1293 1293 .unwrap(); 1294 1294 1295 1295 collection 1296 1296 + } 1297 1297 + pub fn delete_account(&mut self, did: &str, cursor: u64) -> Did { 1298 1298 + let did = Did::new(did.to_string()).unwrap(); 1299 1299 + self.batch.account_removes.push(DeleteAccount { 1300 1300 + did: did.clone(), 1301 1301 + cursor: Cursor::from_raw_u64(cursor), 1302 1302 + }); 1303 1303 + did 1296 1304 } 1297 1305 } 1298 1306 ··· 1532 1540 read.get_records_by_collections(&vec![&Nsid::new("a.a.a".to_string()).unwrap()], 100)?; 1533 1541 assert_eq!(records.len(), 1); 1534 1542 1543 1543 + Ok(()) 1544 1544 + } 1545 1545 + 1546 1546 + #[test] 1547 1547 + fn rollup_delete_live_count_step() -> anyhow::Result<()> { 1548 1548 + let (_read, mut write) = fjall_db(); 1549 1549 + 1550 1550 + let mut batch = TestBatch::default(); 1551 1551 + batch.create( 1552 1552 + "did:plc:person-a", 1553 1553 + "a.a.a", 1554 1554 + "rkey-aaa", 1555 1555 + "{}", 1556 1556 + Some("rev-aaa"), 1557 1557 + None, 1558 1558 + 10_000, 1559 1559 + ); 1560 1560 + write.insert_batch(batch.batch)?; 1561 1561 + 1562 1562 + write.step_rollup()?; 1563 1563 + 1564 1564 + let mut batch = TestBatch::default(); 1565 1565 + batch.delete_account("did:plc:person-a", 10_001); 1566 1566 + write.insert_batch(batch.batch)?; 1567 1567 + 1568 1568 + write.step_rollup()?; 1569 1569 + 1570 1570 + let mut batch = TestBatch::default(); 1571 1571 + batch.delete_account("did:plc:person-a", 9_999); 1572 1572 + write.insert_batch(batch.batch)?; 1573 1573 + 1574 1574 + write.step_rollup()?; 1575 1575 + 1576 1576 + assert!(false); 1577 1577 + Ok(()) 1578 1578 + } 1579 1579 + 1580 1580 + #[test] 1581 1581 + fn rollup_delete_live_count_step_delete_first() -> anyhow::Result<()> { 1582 1582 + let (_read, mut write) = fjall_db(); 1583 1583 + 1584 1584 + let mut batch = TestBatch::default(); 1585 1585 + batch.delete_account("did:plc:person-a", 9_999); 1586 1586 + write.insert_batch(batch.batch)?; 1587 1587 + 1588 1588 + write.step_rollup()?; 1589 1589 + 1590 1590 + assert!(false); 1535 1591 Ok(()) 1536 1592 } 1537 1593 }