Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
apis for ufos demos
author
phil
date
(Apr 14, 2025, 11:46 AM -0400)
commit
3383f8b4
3383f8b43b43ad3e4279beeb43709715f19cb4fe
parent
b04e9ad4
b04e9ad41b92448fd33b64426803ce46f32318e7
+357
-86
+8
-2
ufos/src/consumer.rs
+8
-2
ufos/src/consumer.rs
···
159
159
// holds up all consumer progress until it can send to the channel
160
160
// use this when the current batch is too full to add more to it
161
161
async fn send_current_batch_now(&mut self, small: bool) -> anyhow::Result<()> {
162
+
let beginning = match self.current_batch.initial_cursor.map(|c| c.elapsed()) {
163
+
None => "unknown".to_string(),
164
+
Some(Ok(t)) => format!("{:?}", t),
165
+
Some(Err(e)) => format!("+{:?}", e.duration()),
166
+
};
162
167
log::info!(
163
-
"attempting to send batch now (small? {small}, capacity: {})",
164
-
self.batch_sender.capacity()
168
+
"sending batch now from {beginning}, {}, queue capacity: {}",
169
+
if small { "small" } else { "full" },
170
+
self.batch_sender.capacity(),
165
171
);
166
172
let current = mem::take(&mut self.current_batch);
167
173
self.batch_sender
+47
ufos/src/lib.rs
+47
ufos/src/lib.rs
···
227
227
nsid_child_segments: HashMap<String, TopCollections>,
228
228
}
229
229
230
+
// this is not safe from ~DOS
231
+
// todo: remove this and just iterate the all-time rollups to get nsids? (or recent rollups?)
232
+
impl From<TopCollections> for Vec<String> {
233
+
fn from(tc: TopCollections) -> Self {
234
+
let mut me = vec![];
235
+
for (segment, children) in tc.nsid_child_segments {
236
+
let child_segments: Self = children.into();
237
+
if child_segments.is_empty() {
238
+
me.push(segment);
239
+
} else {
240
+
for ch in child_segments {
241
+
let nsid = format!("{segment}.{ch}");
242
+
me.push(nsid);
243
+
}
244
+
}
245
+
}
246
+
me
247
+
}
248
+
}
249
+
230
250
#[cfg(test)]
231
251
mod tests {
232
252
use super::*;
253
+
254
+
#[test]
255
+
fn test_top_collections_to_nsids() {
256
+
let empty_tc = TopCollections::default();
257
+
assert_eq!(Into::<Vec<String>>::into(empty_tc), Vec::<String>::new());
258
+
259
+
let tc = TopCollections {
260
+
nsid_child_segments: HashMap::from([
261
+
(
262
+
"a".to_string(),
263
+
TopCollections {
264
+
nsid_child_segments: HashMap::from([
265
+
("b".to_string(), TopCollections::default()),
266
+
("c".to_string(), TopCollections::default()),
267
+
]),
268
+
..Default::default()
269
+
},
270
+
),
271
+
("z".to_string(), TopCollections::default()),
272
+
]),
273
+
..Default::default()
274
+
};
275
+
276
+
let mut nsids: Vec<String> = tc.into();
277
+
nsids.sort();
278
+
assert_eq!(nsids, ["a.b", "a.c", "z"]);
279
+
}
233
280
234
281
#[test]
235
282
fn test_truncating_insert_truncates() -> anyhow::Result<()> {
+46
-26
ufos/src/server.rs
+46
-26
ufos/src/server.rs
···
61
61
consumer,
62
62
})
63
63
}
64
+
fn to_multiple_nsids(s: &str) -> Result<Vec<Nsid>, String> {
65
+
let mut out = Vec::new();
66
+
for collection in s.split(',') {
67
+
let Ok(nsid) = Nsid::new(collection.to_string()) else {
68
+
return Err(format!("collection {collection:?} was not a valid NSID"));
69
+
};
70
+
out.push(nsid);
71
+
}
72
+
Ok(out)
73
+
}
64
74
65
75
#[derive(Debug, Deserialize, JsonSchema)]
66
-
struct CollectionsQuery {
67
-
collection: String, // JsonSchema not implemented for Nsid :(
68
-
}
69
-
impl CollectionsQuery {
70
-
fn to_multiple_nsids(&self) -> Result<Vec<Nsid>, String> {
71
-
let mut out = Vec::with_capacity(self.collection.len());
72
-
for collection in self.collection.split(',') {
73
-
let Ok(nsid) = Nsid::new(collection.to_string()) else {
74
-
return Err(format!("collection {collection:?} was not a valid NSID"));
75
-
};
76
-
out.push(nsid);
77
-
}
78
-
Ok(out)
79
-
}
76
+
struct RecordsCollectionsQuery {
77
+
collection: Option<String>, // JsonSchema not implemented for Nsid :(
80
78
}
81
79
#[derive(Debug, Serialize, JsonSchema)]
82
80
struct ApiRecord {
···
105
103
method = GET,
106
104
path = "/records",
107
105
}]
108
-
async fn get_records_by_collection(
106
+
async fn get_records_by_collections(
109
107
ctx: RequestContext<Context>,
110
-
collection_query: Query<CollectionsQuery>,
108
+
collection_query: Query<RecordsCollectionsQuery>,
111
109
) -> OkCorsResponse<Vec<ApiRecord>> {
112
110
let Context { storage, .. } = ctx.context();
111
+
let mut limit = 42;
112
+
let query = collection_query.into_inner();
113
+
let collections = if let Some(provided_collection) = query.collection {
114
+
to_multiple_nsids(&provided_collection)
115
+
.map_err(|reason| HttpError::for_bad_request(None, reason))?
116
+
} else {
117
+
let all_collections_should_be_nsids: Vec<String> = storage
118
+
.get_top_collections()
119
+
.await
120
+
.map_err(|e| {
121
+
HttpError::for_internal_error(format!("failed to get top collections: {e:?}"))
122
+
})?
123
+
.into();
124
+
let mut all_collections = Vec::with_capacity(all_collections_should_be_nsids.len());
125
+
for raw_nsid in all_collections_should_be_nsids {
126
+
let nsid = Nsid::new(raw_nsid).map_err(|e| {
127
+
HttpError::for_internal_error(format!("failed to parse nsid: {e:?}"))
128
+
})?;
129
+
all_collections.push(nsid);
130
+
}
113
131
114
-
let collections = collection_query
115
-
.into_inner()
116
-
.to_multiple_nsids()
117
-
.map_err(|reason| HttpError::for_bad_request(None, reason))?;
132
+
limit = 12;
133
+
all_collections
134
+
};
118
135
119
136
let records = storage
120
-
.get_records_by_collections(&collections, 100)
137
+
.get_records_by_collections(&collections, limit, true)
121
138
.await
122
139
.map_err(|e| HttpError::for_internal_error(e.to_string()))?
123
140
.into_iter()
···
127
144
ok_cors(records)
128
145
}
129
146
147
+
#[derive(Debug, Deserialize, JsonSchema)]
148
+
struct TotalSeenCollectionsQuery {
149
+
collection: String, // JsonSchema not implemented for Nsid :(
150
+
}
130
151
#[derive(Debug, Serialize, JsonSchema)]
131
152
struct TotalCounts {
132
153
total_records: u64,
···
139
160
}]
140
161
async fn get_records_total_seen(
141
162
ctx: RequestContext<Context>,
142
-
collection_query: Query<CollectionsQuery>,
163
+
collection_query: Query<TotalSeenCollectionsQuery>,
143
164
) -> OkCorsResponse<HashMap<String, TotalCounts>> {
144
165
let Context { storage, .. } = ctx.context();
145
166
146
-
let collections = collection_query
147
-
.into_inner()
148
-
.to_multiple_nsids()
167
+
let query = collection_query.into_inner();
168
+
let collections = to_multiple_nsids(&query.collection)
149
169
.map_err(|reason| HttpError::for_bad_request(None, reason))?;
150
170
151
171
let mut seen_by_collection = HashMap::with_capacity(collections.len());
···
194
214
195
215
api.register(get_openapi).unwrap();
196
216
api.register(get_meta_info).unwrap();
197
-
api.register(get_records_by_collection).unwrap();
217
+
api.register(get_records_by_collections).unwrap();
198
218
api.register(get_records_total_seen).unwrap();
199
219
api.register(get_top_collections).unwrap();
200
220
+1
ufos/src/storage.rs
+1
ufos/src/storage.rs
+194
-30
ufos/src/storage_fjall.rs
+194
-30
ufos/src/storage_fjall.rs
···
387
387
&self,
388
388
collections: &[Nsid],
389
389
limit: usize,
390
+
expand_each_collection: bool,
390
391
) -> StorageResult<Vec<UFOsRecord>> {
391
392
if collections.is_empty() {
392
393
return Ok(vec![]);
···
408
409
Err(e) => Err(std::mem::replace(e, StorageError::Stolen))?,
409
410
};
410
411
let Some(rec) = it else {
411
-
break;
412
+
if expand_each_collection {
413
+
continue;
414
+
} else {
415
+
break;
416
+
}
412
417
};
413
418
if let Some((cursor, _)) = latest {
414
419
if rec.cursor > cursor {
···
452
457
&self,
453
458
collections: &[Nsid],
454
459
limit: usize,
460
+
expand_each_collection: bool,
455
461
) -> StorageResult<Vec<UFOsRecord>> {
456
462
let s = self.clone();
457
463
let collections = collections.to_vec();
458
464
tokio::task::spawn_blocking(move || {
459
-
FjallReader::get_records_by_collections(&s, &collections, limit)
465
+
FjallReader::get_records_by_collections(&s, &collections, limit, expand_each_collection)
460
466
})
461
467
.await?
462
468
}
···
578
584
assert_eq!(n, tripppin.len());
579
585
assert_eq!(counts.prefix, and_back.prefix);
580
586
assert_eq!(counts.dids().estimate(), and_back.dids().estimate());
581
-
if counts.records() > 20_000_000 {
587
+
if counts.records() > 200_000_000_000 {
582
588
panic!("COUNTS maybe wtf? {counts:?}")
583
589
}
584
590
···
1066
1072
assert_eq!(records, 0);
1067
1073
assert_eq!(dids, 0);
1068
1074
1069
-
let records = read.get_records_by_collections(&[collection], 2)?;
1075
+
let records = read.get_records_by_collections(&[collection], 2, false)?;
1070
1076
assert_eq!(records.len(), 1);
1071
1077
let rec = &records[0];
1072
1078
assert_eq!(rec.record.get(), "{}");
1073
1079
assert!(!rec.is_update);
1074
1080
1075
1081
let records =
1076
-
read.get_records_by_collections(&[Nsid::new("d.e.f".to_string()).unwrap()], 2)?;
1082
+
read.get_records_by_collections(&[Nsid::new("d.e.f".to_string()).unwrap()], 2, false)?;
1077
1083
assert_eq!(records.len(), 0);
1078
1084
1079
1085
Ok(())
···
1120
1126
Nsid::new("a.a.c".to_string()).unwrap(),
1121
1127
],
1122
1128
100,
1129
+
false,
1123
1130
)?;
1124
1131
assert_eq!(records.len(), 3);
1125
1132
assert_eq!(records[0].record.get(), r#""last""#);
···
1142
1149
}
1143
1150
1144
1151
#[test]
1152
+
fn test_get_multi_collection_expanded() -> anyhow::Result<()> {
1153
+
let (read, mut write) = fjall_db();
1154
+
1155
+
let mut batch = TestBatch::default();
1156
+
// insert some older ones in aab
1157
+
for i in 1..=3 {
1158
+
batch.create(
1159
+
"did:plc:inze6wrmsm7pjl7yta3oig77",
1160
+
"a.a.b",
1161
+
&format!("aab-{i}"),
1162
+
&format!(r#""b {i}""#),
1163
+
Some(&format!("rev-b-{i}")),
1164
+
None,
1165
+
100 + i,
1166
+
);
1167
+
}
1168
+
// and some newer ones in aaa
1169
+
for i in 1..=3 {
1170
+
batch.create(
1171
+
"did:plc:inze6wrmsm7pjl7yta3oig77",
1172
+
"a.a.a",
1173
+
&format!("aaa-{i}"),
1174
+
&format!(r#""a {i}""#),
1175
+
Some(&format!("rev-a-{i}")),
1176
+
None,
1177
+
200 + i,
1178
+
);
1179
+
}
1180
+
write.insert_batch(batch.batch)?;
1181
+
1182
+
let records = read.get_records_by_collections(
1183
+
&[
1184
+
Nsid::new("a.a.a".to_string()).unwrap(),
1185
+
Nsid::new("a.a.b".to_string()).unwrap(),
1186
+
Nsid::new("a.a.c".to_string()).unwrap(),
1187
+
],
1188
+
2,
1189
+
true,
1190
+
)?;
1191
+
assert_eq!(records.len(), 4);
1192
+
assert_eq!(records[0].record.get(), r#""a 3""#);
1193
+
assert_eq!(
1194
+
records[0].collection,
1195
+
Nsid::new("a.a.a".to_string()).unwrap()
1196
+
);
1197
+
1198
+
assert_eq!(records[3].record.get(), r#""b 2""#);
1199
+
assert_eq!(
1200
+
records[3].collection,
1201
+
Nsid::new("a.a.b".to_string()).unwrap()
1202
+
);
1203
+
1204
+
Ok(())
1205
+
}
1206
+
1207
+
#[test]
1145
1208
fn test_update_one() -> anyhow::Result<()> {
1146
1209
let (read, mut write) = fjall_db();
1147
1210
···
1173
1236
assert_eq!(records, 1);
1174
1237
assert_eq!(dids, 1);
1175
1238
1176
-
let records = read.get_records_by_collections(&[collection], 2)?;
1239
+
let records = read.get_records_by_collections(&[collection], 2, false)?;
1177
1240
assert_eq!(records.len(), 1);
1178
1241
let rec = &records[0];
1179
1242
assert_eq!(rec.record.get(), r#"{"ch": "ch-ch-ch-changes"}"#);
···
1211
1274
assert_eq!(records, 1);
1212
1275
assert_eq!(dids, 1);
1213
1276
1214
-
let records = read.get_records_by_collections(&[collection], 2)?;
1277
+
let records = read.get_records_by_collections(&[collection], 2, false)?;
1215
1278
assert_eq!(records.len(), 0);
1216
1279
1217
1280
Ok(())
···
1256
1319
1257
1320
write.insert_batch(batch.batch)?;
1258
1321
1259
-
let records =
1260
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 100)?;
1322
+
let records = read.get_records_by_collections(
1323
+
&[Nsid::new("a.a.a".to_string()).unwrap()],
1324
+
100,
1325
+
false,
1326
+
)?;
1261
1327
assert_eq!(records.len(), 1);
1262
-
let records =
1263
-
read.get_records_by_collections(&[Nsid::new("a.a.b".to_string()).unwrap()], 100)?;
1328
+
let records = read.get_records_by_collections(
1329
+
&[Nsid::new("a.a.b".to_string()).unwrap()],
1330
+
100,
1331
+
false,
1332
+
)?;
1264
1333
assert_eq!(records.len(), 10);
1265
-
let records =
1266
-
read.get_records_by_collections(&[Nsid::new("a.a.c".to_string()).unwrap()], 100)?;
1334
+
let records = read.get_records_by_collections(
1335
+
&[Nsid::new("a.a.c".to_string()).unwrap()],
1336
+
100,
1337
+
false,
1338
+
)?;
1267
1339
assert_eq!(records.len(), 1);
1268
-
let records =
1269
-
read.get_records_by_collections(&[Nsid::new("a.a.d".to_string()).unwrap()], 100)?;
1340
+
let records = read.get_records_by_collections(
1341
+
&[Nsid::new("a.a.d".to_string()).unwrap()],
1342
+
100,
1343
+
false,
1344
+
)?;
1270
1345
assert_eq!(records.len(), 0);
1271
1346
1272
1347
write.trim_collection(&Nsid::new("a.a.a".to_string()).unwrap(), 6)?;
···
1274
1349
write.trim_collection(&Nsid::new("a.a.c".to_string()).unwrap(), 6)?;
1275
1350
write.trim_collection(&Nsid::new("a.a.d".to_string()).unwrap(), 6)?;
1276
1351
1277
-
let records =
1278
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 100)?;
1352
+
let records = read.get_records_by_collections(
1353
+
&[Nsid::new("a.a.a".to_string()).unwrap()],
1354
+
100,
1355
+
false,
1356
+
)?;
1279
1357
assert_eq!(records.len(), 1);
1280
-
let records =
1281
-
read.get_records_by_collections(&[Nsid::new("a.a.b".to_string()).unwrap()], 100)?;
1358
+
let records = read.get_records_by_collections(
1359
+
&[Nsid::new("a.a.b".to_string()).unwrap()],
1360
+
100,
1361
+
false,
1362
+
)?;
1282
1363
assert_eq!(records.len(), 6);
1283
-
let records =
1284
-
read.get_records_by_collections(&[Nsid::new("a.a.c".to_string()).unwrap()], 100)?;
1364
+
let records = read.get_records_by_collections(
1365
+
&[Nsid::new("a.a.c".to_string()).unwrap()],
1366
+
100,
1367
+
false,
1368
+
)?;
1285
1369
assert_eq!(records.len(), 1);
1286
-
let records =
1287
-
read.get_records_by_collections(&[Nsid::new("a.a.d".to_string()).unwrap()], 100)?;
1370
+
let records = read.get_records_by_collections(
1371
+
&[Nsid::new("a.a.d".to_string()).unwrap()],
1372
+
100,
1373
+
false,
1374
+
)?;
1288
1375
assert_eq!(records.len(), 0);
1289
1376
1290
1377
Ok(())
···
1317
1404
}
1318
1405
write.insert_batch(batch.batch)?;
1319
1406
1320
-
let records =
1321
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 100)?;
1407
+
let records = read.get_records_by_collections(
1408
+
&[Nsid::new("a.a.a".to_string()).unwrap()],
1409
+
100,
1410
+
false,
1411
+
)?;
1322
1412
assert_eq!(records.len(), 3);
1323
1413
1324
1414
let records_deleted =
1325
1415
write.delete_account(&Did::new("did:plc:person-b".to_string()).unwrap())?;
1326
1416
assert_eq!(records_deleted, 2);
1327
1417
1328
-
let records =
1329
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 100)?;
1418
+
let records = read.get_records_by_collections(
1419
+
&[Nsid::new("a.a.a".to_string()).unwrap()],
1420
+
100,
1421
+
false,
1422
+
)?;
1330
1423
assert_eq!(records.len(), 1);
1331
1424
1332
1425
Ok(())
···
1355
1448
write.step_rollup()?;
1356
1449
1357
1450
let records =
1358
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1)?;
1451
+
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1, false)?;
1359
1452
assert_eq!(records.len(), 0);
1360
1453
1361
1454
Ok(())
···
1385
1478
write.insert_batch(batch.batch)?;
1386
1479
1387
1480
let records =
1388
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1)?;
1481
+
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1, false)?;
1389
1482
assert_eq!(records.len(), 1);
1390
1483
1391
1484
let n = write.step_rollup()?;
1392
1485
assert_eq!(n, 1);
1393
1486
1394
1487
let records =
1395
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1)?;
1488
+
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1, false)?;
1396
1489
assert_eq!(records.len(), 0);
1397
1490
1398
1491
let mut batch = TestBatch::default();
···
1626
1719
),]),
1627
1720
}
1628
1721
);
1722
+
Ok(())
1723
+
}
1724
+
1725
+
#[test]
1726
+
fn get_top_collections_with_parent_nsid() -> anyhow::Result<()> {
1727
+
let (read, mut write) = fjall_db();
1728
+
1729
+
let mut batch = TestBatch::default();
1730
+
batch.create(
1731
+
"did:plc:inze6wrmsm7pjl7yta3oig77",
1732
+
"a.a.a.a",
1733
+
"aaaa",
1734
+
r#""child nsid""#,
1735
+
Some("rev-aaaa"),
1736
+
None,
1737
+
100,
1738
+
);
1739
+
batch.create(
1740
+
"did:plc:inze6wrmsm7pjl7yta3oig77",
1741
+
"a.a.a",
1742
+
"aaa",
1743
+
r#""parent nsid""#,
1744
+
Some("rev-aaa"),
1745
+
None,
1746
+
101,
1747
+
);
1748
+
write.insert_batch(batch.batch)?;
1749
+
1750
+
let n = write.step_rollup()?;
1751
+
assert_eq!(n, 2); // 3 collections
1752
+
1753
+
let tops = read.get_top_collections()?;
1754
+
assert_eq!(
1755
+
tops,
1756
+
TopCollections {
1757
+
total_records: 2,
1758
+
dids_estimate: 1,
1759
+
nsid_child_segments: HashMap::from([(
1760
+
"a".to_string(),
1761
+
TopCollections {
1762
+
total_records: 2,
1763
+
dids_estimate: 1,
1764
+
nsid_child_segments: HashMap::from([(
1765
+
"a".to_string(),
1766
+
TopCollections {
1767
+
total_records: 2,
1768
+
dids_estimate: 1,
1769
+
nsid_child_segments: HashMap::from([(
1770
+
"a".to_string(),
1771
+
TopCollections {
1772
+
total_records: 2,
1773
+
dids_estimate: 1,
1774
+
nsid_child_segments: HashMap::from([(
1775
+
"a".to_string(),
1776
+
TopCollections {
1777
+
total_records: 1,
1778
+
dids_estimate: 1,
1779
+
nsid_child_segments: HashMap::from([]),
1780
+
},
1781
+
),]),
1782
+
},
1783
+
),]),
1784
+
},
1785
+
),]),
1786
+
},
1787
+
),]),
1788
+
}
1789
+
);
1790
+
1791
+
// TODO: handle leaf node counts explicitly, since parent NSIDs can be leaves themselves
1792
+
1629
1793
Ok(())
1630
1794
}
1631
1795
}
+61
-28
ufos/src/storage_mem.rs
+61
-28
ufos/src/storage_mem.rs
···
528
528
&self,
529
529
collections: &[Nsid],
530
530
limit: usize,
531
+
_expand_each_collection: bool,
531
532
) -> StorageResult<Vec<UFOsRecord>> {
532
533
if collections.is_empty() {
533
534
return Ok(vec![]);
···
593
594
&self,
594
595
collections: &[Nsid],
595
596
limit: usize,
597
+
expand_each_collection: bool,
596
598
) -> StorageResult<Vec<UFOsRecord>> {
597
599
let s = self.clone();
598
600
let collections = collections.to_vec();
599
601
tokio::task::spawn_blocking(move || {
600
-
MemReader::get_records_by_collections(&s, &collections, limit)
602
+
MemReader::get_records_by_collections(&s, &collections, limit, expand_each_collection)
601
603
})
602
604
.await?
603
605
}
···
1243
1245
assert_eq!(records, 0);
1244
1246
assert_eq!(dids, 0);
1245
1247
1246
-
let records = read.get_records_by_collections(&[collection], 2)?;
1248
+
let records = read.get_records_by_collections(&[collection], 2, false)?;
1247
1249
assert_eq!(records.len(), 1);
1248
1250
let rec = &records[0];
1249
1251
assert_eq!(rec.record.get(), "{}");
1250
1252
assert!(!rec.is_update);
1251
1253
1252
1254
let records =
1253
-
read.get_records_by_collections(&[Nsid::new("d.e.f".to_string()).unwrap()], 2)?;
1255
+
read.get_records_by_collections(&[Nsid::new("d.e.f".to_string()).unwrap()], 2, false)?;
1254
1256
assert_eq!(records.len(), 0);
1255
1257
1256
1258
Ok(())
···
1297
1299
Nsid::new("a.a.c".to_string()).unwrap(),
1298
1300
],
1299
1301
100,
1302
+
false,
1300
1303
)?;
1301
1304
assert_eq!(records.len(), 3);
1302
1305
assert_eq!(records[0].record.get(), r#""last""#);
···
1350
1353
assert_eq!(records, 1);
1351
1354
assert_eq!(dids, 1);
1352
1355
1353
-
let records = read.get_records_by_collections(&[collection], 2)?;
1356
+
let records = read.get_records_by_collections(&[collection], 2, false)?;
1354
1357
assert_eq!(records.len(), 1);
1355
1358
let rec = &records[0];
1356
1359
assert_eq!(rec.record.get(), r#"{"ch": "ch-ch-ch-changes"}"#);
···
1388
1391
assert_eq!(records, 1);
1389
1392
assert_eq!(dids, 1);
1390
1393
1391
-
let records = read.get_records_by_collections(&[collection], 2)?;
1394
+
let records = read.get_records_by_collections(&[collection], 2, false)?;
1392
1395
assert_eq!(records.len(), 0);
1393
1396
1394
1397
Ok(())
···
1433
1436
1434
1437
write.insert_batch(batch.batch)?;
1435
1438
1436
-
let records =
1437
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 100)?;
1439
+
let records = read.get_records_by_collections(
1440
+
&[Nsid::new("a.a.a".to_string()).unwrap()],
1441
+
100,
1442
+
false,
1443
+
)?;
1438
1444
assert_eq!(records.len(), 1);
1439
-
let records =
1440
-
read.get_records_by_collections(&[Nsid::new("a.a.b".to_string()).unwrap()], 100)?;
1445
+
let records = read.get_records_by_collections(
1446
+
&[Nsid::new("a.a.b".to_string()).unwrap()],
1447
+
100,
1448
+
false,
1449
+
)?;
1441
1450
assert_eq!(records.len(), 10);
1442
-
let records =
1443
-
read.get_records_by_collections(&[Nsid::new("a.a.c".to_string()).unwrap()], 100)?;
1451
+
let records = read.get_records_by_collections(
1452
+
&[Nsid::new("a.a.c".to_string()).unwrap()],
1453
+
100,
1454
+
false,
1455
+
)?;
1444
1456
assert_eq!(records.len(), 1);
1445
-
let records =
1446
-
read.get_records_by_collections(&[Nsid::new("a.a.d".to_string()).unwrap()], 100)?;
1457
+
let records = read.get_records_by_collections(
1458
+
&[Nsid::new("a.a.d".to_string()).unwrap()],
1459
+
100,
1460
+
false,
1461
+
)?;
1447
1462
assert_eq!(records.len(), 0);
1448
1463
1449
1464
write.trim_collection(&Nsid::new("a.a.a".to_string()).unwrap(), 6)?;
···
1451
1466
write.trim_collection(&Nsid::new("a.a.c".to_string()).unwrap(), 6)?;
1452
1467
write.trim_collection(&Nsid::new("a.a.d".to_string()).unwrap(), 6)?;
1453
1468
1454
-
let records =
1455
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 100)?;
1469
+
let records = read.get_records_by_collections(
1470
+
&[Nsid::new("a.a.a".to_string()).unwrap()],
1471
+
100,
1472
+
false,
1473
+
)?;
1456
1474
assert_eq!(records.len(), 1);
1457
-
let records =
1458
-
read.get_records_by_collections(&[Nsid::new("a.a.b".to_string()).unwrap()], 100)?;
1475
+
let records = read.get_records_by_collections(
1476
+
&[Nsid::new("a.a.b".to_string()).unwrap()],
1477
+
100,
1478
+
false,
1479
+
)?;
1459
1480
assert_eq!(records.len(), 6);
1460
-
let records =
1461
-
read.get_records_by_collections(&[Nsid::new("a.a.c".to_string()).unwrap()], 100)?;
1481
+
let records = read.get_records_by_collections(
1482
+
&[Nsid::new("a.a.c".to_string()).unwrap()],
1483
+
100,
1484
+
false,
1485
+
)?;
1462
1486
assert_eq!(records.len(), 1);
1463
-
let records =
1464
-
read.get_records_by_collections(&[Nsid::new("a.a.d".to_string()).unwrap()], 100)?;
1487
+
let records = read.get_records_by_collections(
1488
+
&[Nsid::new("a.a.d".to_string()).unwrap()],
1489
+
100,
1490
+
false,
1491
+
)?;
1465
1492
assert_eq!(records.len(), 0);
1466
1493
1467
1494
Ok(())
···
1494
1521
}
1495
1522
write.insert_batch(batch.batch)?;
1496
1523
1497
-
let records =
1498
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 100)?;
1524
+
let records = read.get_records_by_collections(
1525
+
&[Nsid::new("a.a.a".to_string()).unwrap()],
1526
+
100,
1527
+
false,
1528
+
)?;
1499
1529
assert_eq!(records.len(), 3);
1500
1530
1501
1531
let records_deleted =
1502
1532
write.delete_account(&Did::new("did:plc:person-b".to_string()).unwrap())?;
1503
1533
assert_eq!(records_deleted, 2);
1504
1534
1505
-
let records =
1506
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 100)?;
1535
+
let records = read.get_records_by_collections(
1536
+
&[Nsid::new("a.a.a".to_string()).unwrap()],
1537
+
100,
1538
+
false,
1539
+
)?;
1507
1540
assert_eq!(records.len(), 1);
1508
1541
1509
1542
Ok(())
···
1532
1565
write.step_rollup()?;
1533
1566
1534
1567
let records =
1535
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1)?;
1568
+
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1, false)?;
1536
1569
assert_eq!(records.len(), 0);
1537
1570
1538
1571
Ok(())
···
1562
1595
write.insert_batch(batch.batch)?;
1563
1596
1564
1597
let records =
1565
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1)?;
1598
+
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1, false)?;
1566
1599
assert_eq!(records.len(), 1);
1567
1600
1568
1601
let n = write.step_rollup()?;
1569
1602
assert_eq!(n, 1);
1570
1603
1571
1604
let records =
1572
-
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1)?;
1605
+
read.get_records_by_collections(&[Nsid::new("a.a.a".to_string()).unwrap()], 1, false)?;
1573
1606
assert_eq!(records.len(), 0);
1574
1607
1575
1608
let mut batch = TestBatch::default();