+17 -1

ufos/src/server.rs

Reviewed

··· 213 213 ok_cors(seen_by_collection) 214 214 } 215 215 216 216 - /// Get top collections by count 216 216 + /// Get top collections by record count 217 217 #[endpoint { 218 218 method = GET, 219 219 path = "/collections/by-count" ··· 222 222 let Context { storage, .. } = ctx.context(); 223 223 let collections = storage 224 224 .get_top_collections_by_count(100, QueryPeriod::all_time()) 225 225 + .await 226 226 + .map_err(|e| HttpError::for_internal_error(format!("oh shoot: {e:?}")))?; 227 227 + 228 228 + ok_cors(collections) 229 229 + } 230 230 + 231 231 + /// Get top collections by estimated unique DIDs 232 232 + #[endpoint { 233 233 + method = GET, 234 234 + path = "/collections/by-dids" 235 235 + }] 236 236 + async fn get_top_collections_by_dids(ctx: RequestContext<Context>) -> OkCorsResponse<Vec<Count>> { 237 237 + let Context { storage, .. } = ctx.context(); 238 238 + let collections = storage 239 239 + .get_top_collections_by_dids(100, QueryPeriod::all_time()) 225 240 .await 226 241 .map_err(|e| HttpError::for_internal_error(format!("oh shoot: {e:?}")))?; 227 242 ··· 260 275 api.register(get_records_by_collections).unwrap(); 261 276 api.register(get_records_total_seen).unwrap(); 262 277 api.register(get_top_collections_by_count).unwrap(); 278 278 + api.register(get_top_collections_by_dids).unwrap(); 263 279 api.register(get_top_collections).unwrap(); 264 280 265 281 let context = Context {

+6

ufos/src/storage.rs

Reviewed

··· 76 76 period: QueryPeriod, 77 77 ) -> StorageResult<Vec<Count>>; 78 78 79 79 + async fn get_top_collections_by_dids( 80 80 + &self, 81 81 + limit: usize, 82 82 + period: QueryPeriod, 83 83 + ) -> StorageResult<Vec<Count>>; 84 84 + 79 85 async fn get_top_collections(&self) -> StorageResult<TopCollections>; 80 86 81 87 async fn get_counts_by_collection(&self, collection: &Nsid) -> StorageResult<(u64, u64)>;

+42 -1

ufos/src/storage_fjall.rs

Reviewed

··· 359 359 "integrity: all-time rank rollup must have corresponding all-time count rollup", 360 360 ); 361 361 let db_counts = db_complete::<CountsValue>(&db_count_bytes)?; 362 362 - assert_eq!(db_counts.records(), key.records()); 362 362 + assert_eq!(db_counts.records(), key.count()); 363 363 + out.push(Count { 364 364 + thing: key.collection().to_string(), 365 365 + records: db_counts.records(), 366 366 + dids_estimate: db_counts.dids().estimate() as u64, 367 367 + }); 368 368 + } 369 369 + out 370 370 + } else { 371 371 + todo!() 372 372 + }) 373 373 + } 374 374 + 375 375 + fn get_top_collections_by_dids( 376 376 + &self, 377 377 + limit: usize, 378 378 + period: QueryPeriod, 379 379 + ) -> StorageResult<Vec<Count>> { 380 380 + Ok(if period.is_all_time() { 381 381 + let snapshot = self.rollups.snapshot(); 382 382 + let mut out = Vec::with_capacity(limit); 383 383 + let prefix = AllTimeDidsKey::from_prefix_to_db_bytes(&Default::default())?; 384 384 + for kv in snapshot.prefix(prefix).rev().take(limit) { 385 385 + let (key_bytes, _) = kv?; 386 386 + let key = db_complete::<AllTimeDidsKey>(&key_bytes)?; 387 387 + let rollup_key = AllTimeRollupKey::new(key.collection()); 388 388 + let db_count_bytes = snapshot.get(rollup_key.to_db_bytes()?)?.expect( 389 389 + "integrity: all-time rank rollup must have corresponding all-time count rollup", 390 390 + ); 391 391 + let db_counts = db_complete::<CountsValue>(&db_count_bytes)?; 392 392 + assert_eq!(db_counts.dids().estimate() as u64, key.count()); 363 393 out.push(Count { 364 394 thing: key.collection().to_string(), 365 395 records: db_counts.records(), ··· 521 551 let s = self.clone(); 522 552 tokio::task::spawn_blocking(move || { 523 553 FjallReader::get_top_collections_by_count(&s, limit, period) 554 554 + }) 555 555 + .await? 556 556 + } 557 557 + async fn get_top_collections_by_dids( 558 558 + &self, 559 559 + limit: usize, 560 560 + period: QueryPeriod, 561 561 + ) -> StorageResult<Vec<Count>> { 562 562 + let s = self.clone(); 563 563 + tokio::task::spawn_blocking(move || { 564 564 + FjallReader::get_top_collections_by_dids(&s, limit, period) 524 565 }) 525 566 .await? 526 567 }

+7

ufos/src/storage_mem.rs

Reviewed

··· 595 595 ) -> StorageResult<Vec<Count>> { 596 596 todo!() 597 597 } 598 598 + async fn get_top_collections_by_dids( 599 599 + &self, 600 600 + _: usize, 601 601 + _: QueryPeriod, 602 602 + ) -> StorageResult<Vec<Count>> { 603 603 + todo!() 604 604 + } 598 605 async fn get_counts_by_collection(&self, collection: &Nsid) -> StorageResult<(u64, u64)> { 599 606 let s = self.clone(); 600 607 let collection = collection.clone();

+1 -1

ufos/src/store_types.rs

Reviewed

··· 366 366 pub fn with_rank(&self, new_rank: KeyRank) -> Self { 367 367 Self::new(new_rank, &self.suffix.suffix) 368 368 } 369 369 - pub fn records(&self) -> u64 { 369 369 + pub fn count(&self) -> u64 { 370 370 self.suffix.prefix.0 371 371 } 372 372 pub fn collection(&self) -> &Nsid {