tracks lexicons and how many times they appeared on the jetstream
fix(server): tracking_since funcs, also fix old db get_hits
+57
-43
+57
-43
server/src/db/mod.rs
+57
-43
server/src/db/mod.rs
···
106
106
}
107
107
108
108
#[inline(always)]
109
+
fn get_part_opts() -> PartitionCreateOptions {
110
+
PartitionCreateOptions::default()
111
+
.compression(fjall::CompressionType::Miniz(9))
112
+
.compaction_strategy(fjall::compaction::Strategy::Fifo(fjall::compaction::Fifo {
113
+
limit: 5 * 1024 * 1024 * 1024, // 5 gb
114
+
ttl_seconds: Some(60 * 60 * 24 * 30), // 30 days
115
+
}))
116
+
}
117
+
118
+
#[inline(always)]
119
+
fn maybe_run_in_nsid_tree<T>(&self, nsid: &str, f: impl FnOnce(&Partition) -> T) -> Option<T> {
120
+
let _guard = scc::ebr::Guard::new();
121
+
let handle = match self.hits.peek(nsid, &_guard) {
122
+
Some(handle) => handle.clone(),
123
+
None => {
124
+
if self.inner.partition_exists(nsid) {
125
+
let handle = self
126
+
.inner
127
+
.open_partition(nsid, Self::get_part_opts())
128
+
.expect("cant open partition");
129
+
let _ = self.hits.insert(SmolStr::new(nsid), handle.clone());
130
+
handle
131
+
} else {
132
+
return None;
133
+
}
134
+
}
135
+
};
136
+
Some(f(&handle))
137
+
}
138
+
139
+
#[inline(always)]
109
140
fn run_in_nsid_tree<T>(
110
141
&self,
111
142
nsid: &str,
···
115
146
.hits
116
147
.entry(SmolStr::new(nsid))
117
148
.or_insert_with(|| {
118
-
let opts = PartitionCreateOptions::default()
119
-
.compression(fjall::CompressionType::Miniz(9))
120
-
.compaction_strategy(fjall::compaction::Strategy::Fifo(
121
-
fjall::compaction::Fifo {
122
-
limit: 5 * 1024 * 1024 * 1024, // 5 gb
123
-
ttl_seconds: Some(60 * 60 * 24 * 30), // 30 days
124
-
},
125
-
));
149
+
let opts = Self::get_part_opts();
126
150
self.inner.open_partition(nsid, opts).unwrap()
127
151
})
128
152
.get())
···
203
227
&self,
204
228
nsid: &str,
205
229
range: impl RangeBounds<u64>,
206
-
) -> AppResult<Box<dyn Iterator<Item = AppResult<(u64, NsidHit)>>>> {
230
+
) -> BoxedIter<AppResult<(u64, NsidHit)>> {
207
231
let start = range.start_bound().cloned().map(u64::to_be_bytes);
208
232
let end = range.end_bound().cloned().map(u64::to_be_bytes);
209
233
210
-
let _guard = scc::ebr::Guard::new();
211
-
let Some(tree) = self.hits.peek(nsid, &_guard) else {
212
-
return Ok(Box::new(std::iter::empty()));
213
-
};
214
-
215
-
Ok(Box::new(tree.range(TimestampRangeOld { start, end }).map(
216
-
|res| {
234
+
self.maybe_run_in_nsid_tree(nsid, |tree| -> BoxedIter<AppResult<(u64, NsidHit)>> {
235
+
Box::new(tree.range(TimestampRangeOld { start, end }).map(|res| {
217
236
res.map_err(AppError::from).map(|(key, val)| {
218
237
(
219
238
u64::from_be_bytes(key.as_ref().try_into().unwrap()),
220
239
unsafe { rkyv::from_bytes_unchecked::<_, Error>(&val).unwrap_unchecked() },
221
240
)
222
241
})
223
-
},
224
-
)))
242
+
}))
243
+
})
244
+
.unwrap_or_else(|| Box::new(std::iter::empty()))
225
245
}
226
246
227
247
pub fn tracking_since(&self) -> AppResult<u64> {
228
-
let _guard = scc::ebr::Guard::new();
229
248
// HACK: we should actually store when we started tracking but im lazy
230
249
// should be accurate enough
231
-
let Some(tree) = self.hits.peek("app.bsky.feed.like", &_guard) else {
232
-
return Ok(0);
233
-
};
234
-
let Some((timestamp_raw, _)) = tree.first_key_value()? else {
235
-
return Ok(0);
236
-
};
237
-
drop(_guard);
238
-
239
-
Ok(u64::from_be_bytes(
240
-
timestamp_raw.as_ref().try_into().unwrap(),
241
-
))
250
+
self.maybe_run_in_nsid_tree("app.bsky.feed.like", |tree| {
251
+
let Some((timestamp_raw, _)) = tree.first_key_value()? else {
252
+
return Ok(0);
253
+
};
254
+
Ok(u64::from_be_bytes(
255
+
timestamp_raw.as_ref().try_into().unwrap(),
256
+
))
257
+
})
258
+
.unwrap_or(Ok(0))
242
259
}
243
260
}
244
261
···
532
549
}
533
550
534
551
pub fn tracking_since(&self) -> AppResult<u64> {
535
-
let _guard = scc::ebr::Guard::new();
536
552
// HACK: we should actually store when we started tracking but im lazy
537
553
// should be accurate enough
538
-
let Some(handle) = self.hits.peek("app.bsky.feed.like", &_guard) else {
539
-
return Ok(0);
540
-
};
541
-
let Some((timestamps_raw, _)) = handle.tree.first_key_value()? else {
542
-
return Ok(0);
543
-
};
544
-
drop(_guard);
545
-
546
-
let mut timestamp_reader = Cursor::new(timestamps_raw);
547
-
timestamp_reader
548
-
.read_varint::<u64>()
549
-
.map_err(AppError::from)
554
+
self.maybe_run_in_nsid_tree("app.bsky.feed.like", |handle| {
555
+
let Some((timestamps_raw, _)) = handle.tree.first_key_value()? else {
556
+
return Ok(0);
557
+
};
558
+
let mut timestamp_reader = Cursor::new(timestamps_raw);
559
+
timestamp_reader
560
+
.read_varint::<u64>()
561
+
.map_err(AppError::from)
562
+
})
563
+
.unwrap_or(Ok(0))
550
564
}
551
565
}
552
566