very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
rust
fjall
at-protocol
atproto
indexer
[db] simplify v1 migration
+6
-13
+6
-13
src/db/migration/v1.rs
+6
-13
src/db/migration/v1.rs
···
8
8
9
9
/// migrates firehose cursors from `firehose_cursor|{url}` to `firehose_cursor|{host}`.
10
10
pub(super) fn stable_firehose_cursors(db: &Db, batch: &mut OwnedWriteBatch) -> Result<()> {
11
-
let entries: Vec<(Vec<u8>, Vec<u8>)> = db
12
-
.cursors
13
-
.prefix(keys::FIREHOSE_CURSOR_PREFIX)
14
-
.map(|item| {
15
-
let (k, v) = item.into_inner().into_diagnostic()?;
16
-
Ok((k.to_vec(), v.to_vec()))
17
-
})
18
-
.collect::<Result<_>>()?;
19
-
20
-
for (old_key, value) in entries {
21
-
let suffix = &old_key[keys::FIREHOSE_CURSOR_PREFIX.len()..];
11
+
let prefix = keys::FIREHOSE_CURSOR_PREFIX;
12
+
for item in db.cursors.prefix(prefix) {
13
+
let (old_key, value) = item.into_inner().into_diagnostic()?;
14
+
let suffix = &old_key[prefix.len()..];
22
15
// old-format: suffix is a full URL containing "://" (e.g. "wss://bsky.network")
23
16
// new-format (v1): suffix is just a hostname, no "://"
24
17
if !suffix.windows(3).any(|w| w == b"://") {
···
32
25
.wrap_err_with(|| format!("firehose cursor key contains invalid url {url_str:?}"))?;
33
26
34
27
let new_key = keys::v1::firehose_cursor_key_from_url(&url);
35
-
batch.insert(&db.cursors, &new_key, &value);
36
-
batch.remove(&db.cursors, &old_key);
28
+
batch.insert(&db.cursors, &new_key, value);
29
+
batch.remove(&db.cursors, old_key);
37
30
}
38
31
39
32
Ok(())