+29
-3
ufos/src/lib.rs
+29
-3
ufos/src/lib.rs
···
10
10
11
11
use crate::db_types::{EncodingError, EncodingResult};
12
12
use crate::error::BatchInsertError;
13
-
use crate::store_types::SketchSecretPrefix;
13
+
use crate::store_types::{CountsValue, SketchSecretPrefix};
14
14
use cardinality_estimator_safe::{Element, Sketch};
15
15
use error::FirehoseEventError;
16
16
use jetstream::events::{CommitEvent, CommitOp, Cursor};
···
281
281
pub struct NsidCount {
282
282
nsid: String,
283
283
creates: u64,
284
-
// TODO: add updates and deletes
284
+
updates: u64,
285
+
deletes: u64,
285
286
dids_estimate: u64,
287
+
}
288
+
impl NsidCount {
289
+
pub fn new(nsid: &Nsid, counts: &CountsValue) -> Self {
290
+
let crud = counts.counts();
291
+
Self {
292
+
nsid: nsid.to_string(),
293
+
creates: crud.creates,
294
+
updates: crud.updates,
295
+
deletes: crud.deletes,
296
+
dids_estimate: counts.dids().estimate() as u64,
297
+
}
298
+
}
286
299
}
287
300
288
301
#[derive(Debug, PartialEq, Serialize, JsonSchema)]
289
302
pub struct PrefixCount {
290
303
prefix: String,
291
304
creates: u64,
292
-
// TODO: add updates and deletes
305
+
updates: u64,
306
+
deletes: u64,
293
307
dids_estimate: u64,
308
+
}
309
+
impl PrefixCount {
310
+
pub fn new(prefix: &str, counts: &CountsValue) -> Self {
311
+
let crud = counts.counts();
312
+
Self {
313
+
prefix: prefix.to_string(),
314
+
creates: crud.creates,
315
+
updates: crud.updates,
316
+
deletes: crud.deletes,
317
+
dids_estimate: counts.dids().estimate() as u64,
318
+
}
319
+
}
294
320
}
295
321
296
322
#[derive(Debug, PartialEq, Serialize, JsonSchema)]
+27
-35
ufos/src/storage_fjall.rs
+27
-35
ufos/src/storage_fjall.rs
···
507
507
merged.merge(&counts);
508
508
}
509
509
}
510
-
out.push(NsidCount {
511
-
nsid: nsid.to_string(),
512
-
creates: merged.counts().creates,
513
-
dids_estimate: merged.dids().estimate() as u64,
514
-
});
510
+
out.push(NsidCount::new(&nsid, &merged));
515
511
}
516
512
517
513
let next_cursor = current_nsid.map(|s| s.to_db_bytes()).transpose()?;
···
617
613
.into_iter()
618
614
.rev()
619
615
.take(limit)
620
-
.map(|(nsid, cv)| NsidCount {
621
-
nsid: nsid.to_string(),
622
-
creates: cv.counts().creates,
623
-
dids_estimate: cv.dids().estimate() as u64,
624
-
})
616
+
.map(|(nsid, cv)| NsidCount::new(&nsid, &cv))
625
617
.collect();
626
618
Ok(counts)
627
619
}
···
727
719
let mut prefix_count = CountsValue::default();
728
720
#[derive(Debug, Clone, PartialEq)]
729
721
enum Child {
730
-
FullNsid(String),
722
+
FullNsid(Nsid),
731
723
ChildPrefix(String),
732
724
}
733
725
impl Child {
734
726
fn from_prefix(nsid: &Nsid, prefix: &NsidPrefix) -> Option<Self> {
735
727
if prefix.is_group_of(nsid) {
736
-
return Some(Child::FullNsid(nsid.to_string()));
728
+
return Some(Child::FullNsid(nsid.clone()));
737
729
}
738
730
let suffix = nsid.as_str().strip_prefix(&format!("{}.", prefix.0))?;
739
731
let (segment, _) = suffix.split_once('.').unwrap();
···
742
734
}
743
735
fn is_before(&self, other: &Child) -> bool {
744
736
match (self, other) {
745
-
(Child::FullNsid(s), Child::ChildPrefix(o)) if s == o => true,
746
-
(Child::ChildPrefix(s), Child::FullNsid(o)) if s == o => false,
747
-
(Child::FullNsid(s), Child::FullNsid(o)) => s < o,
737
+
(Child::FullNsid(s), Child::ChildPrefix(o)) if s.as_str() == o => true,
738
+
(Child::ChildPrefix(s), Child::FullNsid(o)) if s == o.as_str() => false,
739
+
(Child::FullNsid(s), Child::FullNsid(o)) => s.as_str() < o.as_str(),
748
740
(Child::ChildPrefix(s), Child::ChildPrefix(o)) => s < o,
749
-
(Child::FullNsid(s), Child::ChildPrefix(o)) => s < o,
750
-
(Child::ChildPrefix(s), Child::FullNsid(o)) => s < o,
741
+
(Child::FullNsid(s), Child::ChildPrefix(o)) => s.to_string() < *o,
742
+
(Child::ChildPrefix(s), Child::FullNsid(o)) => *s < o.to_string(),
751
743
}
752
744
}
753
745
fn into_inner(self) -> String {
754
746
match self {
755
-
Child::FullNsid(s) => s,
747
+
Child::FullNsid(s) => s.to_string(),
756
748
Child::ChildPrefix(s) => s,
757
749
}
758
750
}
···
791
783
}
792
784
}
793
785
items.push(match child {
794
-
Child::FullNsid(nsid) => PrefixChild::Collection(NsidCount {
795
-
nsid,
796
-
creates: merged.counts().creates,
797
-
dids_estimate: merged.dids().estimate() as u64,
798
-
}),
799
-
Child::ChildPrefix(prefix) => PrefixChild::Prefix(PrefixCount {
800
-
prefix,
801
-
creates: merged.counts().creates,
802
-
dids_estimate: merged.dids().estimate() as u64,
803
-
}),
786
+
Child::FullNsid(nsid) => PrefixChild::Collection(NsidCount::new(&nsid, &merged)),
787
+
Child::ChildPrefix(prefix) => {
788
+
PrefixChild::Prefix(PrefixCount::new(&prefix, &merged))
789
+
}
804
790
});
805
791
}
806
792
···
991
977
for kv in self.rollups.range((start, end)) {
992
978
let (key_bytes, val_bytes) = kv?;
993
979
let key = db_complete::<AllTimeRollupKey>(&key_bytes)?;
994
-
let nsid = key.collection().as_str().to_string();
980
+
let nsid = key.collection();
995
981
for term in &terms {
996
982
if nsid.contains(term) {
997
983
let counts = db_complete::<CountsValue>(&val_bytes)?;
998
-
matches.push(NsidCount {
999
-
nsid: nsid.clone(),
1000
-
creates: counts.counts().creates,
1001
-
dids_estimate: counts.dids().estimate() as u64,
1002
-
});
984
+
matches.push(NsidCount::new(nsid, &counts));
1003
985
break;
1004
986
}
1005
987
}
···
2649
2631
vec![PrefixChild::Collection(NsidCount {
2650
2632
nsid: "a.a.a".to_string(),
2651
2633
creates: 1,
2634
+
updates: 0,
2635
+
deletes: 0,
2652
2636
dids_estimate: 1
2653
2637
}),]
2654
2638
);
···
2695
2679
vec![PrefixChild::Prefix(PrefixCount {
2696
2680
prefix: "a.a.a".to_string(),
2697
2681
creates: 1,
2698
-
dids_estimate: 1
2682
+
updates: 0,
2683
+
deletes: 0,
2684
+
dids_estimate: 1,
2699
2685
}),]
2700
2686
);
2701
2687
assert_eq!(cursor, None);
···
2750
2736
vec![PrefixChild::Prefix(PrefixCount {
2751
2737
prefix: "a.a.a".to_string(),
2752
2738
creates: 2,
2739
+
updates: 0,
2740
+
deletes: 0,
2753
2741
dids_estimate: 1
2754
2742
}),]
2755
2743
);
···
2818
2806
PrefixChild::Collection(NsidCount {
2819
2807
nsid: "a.a.a.a".to_string(),
2820
2808
creates: 1,
2809
+
updates: 0,
2810
+
deletes: 0,
2821
2811
dids_estimate: 1
2822
2812
}),
2823
2813
PrefixChild::Prefix(PrefixCount {
2824
2814
prefix: "a.a.a.a".to_string(),
2825
2815
creates: 1,
2816
+
updates: 0,
2817
+
deletes: 0,
2826
2818
dids_estimate: 1
2827
2819
}),
2828
2820
]