forked from
microcosm.blue/microcosm-rs
Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
min hourly/weekly buckets spanning hours
+133
-4
+12
-1
ufos/src/server.rs
+12
-1
ufos/src/server.rs
···
21
21
use serde::{Deserialize, Serialize};
22
22
use std::collections::HashMap;
23
23
use std::sync::Arc;
24
+
use std::time::{SystemTime, UNIX_EPOCH};
24
25
25
26
struct Context {
26
27
pub spec: Arc<serde_json::Value>,
···
32
33
if t < 0 {
33
34
Err(HttpError::for_bad_request(None, "timestamp too old".into()))
34
35
} else {
35
-
Ok(HourTruncatedCursor::truncate_raw_u64(t as u64))
36
+
let t = t as u64;
37
+
let t_now = SystemTime::now()
38
+
.duration_since(UNIX_EPOCH)
39
+
.unwrap()
40
+
.as_micros() as u64;
41
+
const ONE_HOUR: u64 = 60 * 60 * 1_000_000;
42
+
if t < t_now || (t - t_now <= 2 * ONE_HOUR) {
43
+
Err(HttpError::for_bad_request(None, "future timestamp".into()))
44
+
} else {
45
+
Ok(HourTruncatedCursor::truncate_raw_u64(t))
46
+
}
36
47
}
37
48
}
38
49
+121
-3
ufos/src/store_types.rs
+121
-3
ufos/src/store_types.rs
···
387
387
#[derive(Debug, Copy, Clone, PartialEq, Hash, PartialOrd, Eq)]
388
388
pub struct TruncatedCursor<const MOD: u64>(u64);
389
389
impl<const MOD: u64> TruncatedCursor<MOD> {
390
-
fn truncate(raw: u64) -> u64 {
390
+
pub fn truncate(raw: u64) -> u64 {
391
391
(raw / MOD) * MOD
392
392
}
393
393
pub fn try_from_raw_u64(time_us: u64) -> Result<Self, EncodingError> {
···
409
409
let truncated = Self::truncate(raw);
410
410
Self(truncated)
411
411
}
412
+
pub fn to_raw_u64(&self) -> u64 {
413
+
self.0
414
+
}
415
+
pub fn try_as<const MOD_B: u64>(&self) -> Result<TruncatedCursor<MOD_B>, EncodingError> {
416
+
TruncatedCursor::<MOD_B>::try_from_raw_u64(self.0)
417
+
}
418
+
pub fn cycles_until(&self, other: Self) -> u64 {
419
+
if other < *self {
420
+
panic!("other must be greater than or equal to self");
421
+
}
422
+
(other.0 - self.0) / MOD
423
+
}
424
+
pub fn next(&self) -> Self {
425
+
Self(self.0 + MOD)
426
+
}
412
427
}
413
428
impl<const MOD: u64> From<TruncatedCursor<MOD>> for Cursor {
414
429
fn from(truncated: TruncatedCursor<MOD>) -> Self {
···
438
453
const WEEK_IN_MICROS: u64 = HOUR_IN_MICROS * 24 * 7;
439
454
pub type WeekTruncatedCursor = TruncatedCursor<WEEK_IN_MICROS>;
440
455
456
+
#[derive(Debug, PartialEq)]
457
+
pub enum CursorBucket {
458
+
Hour(HourTruncatedCursor),
459
+
Week(WeekTruncatedCursor),
460
+
}
461
+
462
+
impl CursorBucket {
463
+
pub fn buckets_spanning(
464
+
since: HourTruncatedCursor,
465
+
until: HourTruncatedCursor,
466
+
) -> Vec<CursorBucket> {
467
+
if until <= since {
468
+
return vec![];
469
+
}
470
+
let mut out = vec![];
471
+
let mut current_lower = since;
472
+
while current_lower < until {
473
+
if current_lower.cycles_until(until) >= (WEEK_IN_MICROS / HOUR_IN_MICROS) {
474
+
if let Ok(week) = current_lower.try_as::<WEEK_IN_MICROS>() {
475
+
out.push(CursorBucket::Week(week));
476
+
current_lower = week.next().try_as().unwrap();
477
+
continue;
478
+
}
479
+
}
480
+
out.push(CursorBucket::Hour(current_lower));
481
+
current_lower = current_lower.next();
482
+
}
483
+
out
484
+
}
485
+
}
486
+
441
487
#[cfg(test)]
442
488
mod test {
443
489
use super::{
444
-
CountsValue, Cursor, Did, EncodingError, HourTruncatedCursor, HourlyRollupKey, Nsid,
445
-
Sketch, HOUR_IN_MICROS,
490
+
CountsValue, Cursor, CursorBucket, Did, EncodingError, HourTruncatedCursor,
491
+
HourlyRollupKey, Nsid, Sketch, HOUR_IN_MICROS, WEEK_IN_MICROS,
446
492
};
447
493
use crate::db_types::DbBytes;
448
494
use cardinality_estimator_safe::Element;
···
513
559
assert_eq!(back, us);
514
560
let diff = us.to_raw_u64() - back.to_raw_u64();
515
561
assert_eq!(diff, 0);
562
+
}
563
+
564
+
#[test]
565
+
fn test_spanning_nothing() {
566
+
let from = Cursor::from_raw_u64(1_743_775_200_000_000).into();
567
+
let until = Cursor::from_raw_u64(1_743_775_200_000_000).into();
568
+
assert!(CursorBucket::buckets_spanning(from, until).is_empty());
569
+
let until = Cursor::from_raw_u64(0).into();
570
+
assert!(CursorBucket::buckets_spanning(from, until).is_empty());
571
+
}
572
+
573
+
#[test]
574
+
fn test_spanning_low_hours() {
575
+
let from = HourTruncatedCursor::truncate_cursor(Cursor::from_start());
576
+
let until = from.next();
577
+
assert_eq!(
578
+
CursorBucket::buckets_spanning(from, until),
579
+
vec![CursorBucket::Hour(from)]
580
+
);
581
+
let until2 = until.next();
582
+
let until3 = until2.next();
583
+
assert_eq!(
584
+
CursorBucket::buckets_spanning(from, until3),
585
+
vec![
586
+
CursorBucket::Hour(from),
587
+
CursorBucket::Hour(until),
588
+
CursorBucket::Hour(until2),
589
+
]
590
+
);
591
+
}
592
+
593
+
#[test]
594
+
fn test_spanning_week_aligned() {
595
+
let from = HourTruncatedCursor::truncate_cursor(Cursor::from_start());
596
+
let until = HourTruncatedCursor::truncate_cursor(Cursor::from_raw_u64(WEEK_IN_MICROS));
597
+
assert_eq!(
598
+
CursorBucket::buckets_spanning(from, until),
599
+
vec![CursorBucket::Week(from.try_as().unwrap()),]
600
+
);
601
+
let next_hour = until.next();
602
+
assert_eq!(
603
+
CursorBucket::buckets_spanning(from, next_hour),
604
+
vec![
605
+
CursorBucket::Week(from.try_as().unwrap()),
606
+
CursorBucket::Hour(until),
607
+
]
608
+
);
609
+
}
610
+
611
+
#[test]
612
+
fn test_spanning_week_unaligned() {
613
+
let from = HourTruncatedCursor::truncate_cursor(Cursor::from_raw_u64(
614
+
WEEK_IN_MICROS - HOUR_IN_MICROS,
615
+
));
616
+
let until = HourTruncatedCursor::truncate_cursor(Cursor::from_raw_u64(
617
+
from.to_raw_u64() + WEEK_IN_MICROS,
618
+
));
619
+
let span = CursorBucket::buckets_spanning(from, until);
620
+
assert_eq!(span.len(), 168);
621
+
for b in &span {
622
+
let CursorBucket::Hour(_) = b else {
623
+
panic!("found week bucket in a span that should only have hourlies");
624
+
};
625
+
}
626
+
let until2 = until.next();
627
+
assert_eq!(
628
+
CursorBucket::buckets_spanning(from, until2),
629
+
vec![
630
+
CursorBucket::Hour(from),
631
+
CursorBucket::Week(from.next().try_as().unwrap()),
632
+
]
633
+
);
516
634
}
517
635
}