use jacquard_common::types::string::Did;
use miette::{IntoDiagnostic, WrapErr};
use super::SEP;
use crate::db::types::{DbRkey, DbTid, TrimmedDid};
#[cfg(feature = "indexer_stream")]
pub const EVENT_WATERMARK_PREFIX: &[u8] = b"ewm|";
pub fn pending_key(id: u64) -> [u8; 8] {
id.to_be_bytes()
}
// key format: {DID} 00 {ID} (DID trimmed). the NUL keeps did order, because a
// did only prefixes another when both are text, and text never holds a NUL.
// plc keys are binary and can, so take these apart with `split_pending_did_key`
pub fn pending_did_key(repo_key: &[u8], pending_key: &[u8]) -> Vec<u8> {
let mut key = Vec::with_capacity(repo_key.len() + 1 + pending_key.len());
key.extend_from_slice(repo_key);
key.push(REC_SEP);
key.extend_from_slice(pending_key);
key
}
/// the repo key and pending key a [`pending_did_key`] was built from. split
/// from the end, since a plc repo key is binary and may hold a NUL itself
pub fn split_pending_did_key(key: &[u8]) -> Option<(&[u8], &[u8])> {
let sep = key.len().checked_sub(9)?;
(key[sep] == REC_SEP).then(|| (&key[..sep], &key[sep + 1..]))
}
#[cfg(feature = "indexer_stream")]
pub fn event_watermark_key(timestamp_secs: u64) -> Vec<u8> {
let mut key = Vec::with_capacity(EVENT_WATERMARK_PREFIX.len() + 8);
key.extend_from_slice(EVENT_WATERMARK_PREFIX);
key.extend_from_slice(×tamp_secs.to_be_bytes());
key
}
/// segment separator in record keys: NUL terminates the trimmed DID, and in
/// history keys separates the record key from the death revision. NUL is
/// outside every segment's charset (trimmed DID bytes, NSID collections, and
/// rkey text), so it splits unambiguously.
pub const REC_SEP: u8 = 0x00;
/// length of the trimmed-DID segment at the start of a record or history key,
/// for both the current (NUL-terminated) and legacy (`|`-terminated) formats.
///
/// this cannot be a scan for NUL: plc segments are raw binary and may contain
/// NUL bytes, so the tag byte decides. web/other segments are text and cannot
/// contain NUL or `|`.
pub fn trimmed_did_len(key: &[u8]) -> Option<usize> {
use crate::db::types::{TAG_PLC, TAG_WEB};
match key.first()? {
&TAG_PLC => (key.len() > 16).then_some(16),
&TAG_WEB => key[1..]
.iter()
.position(|&b| b == REC_SEP || b == SEP)
.map(|i| i + 1),
_ => key.iter().position(|&b| b == REC_SEP || b == SEP),
}
}
/// collection/rkey separator inside record keys. `/` is in neither charset,
/// and makes the key body read as `{collection}/{rkey}`, which sorts in
/// exactly MST order: TIDs and string rkeys interleave lexicographically,
/// and a record always sorts before records it prefix-matches.
pub const REC_PATH_SEP: u8 = b'/';
// prefix format: {DID} 00 (DID trimmed)
pub fn record_prefix_trimmed(did: &TrimmedDid) -> Vec<u8> {
let mut prefix = Vec::with_capacity(did.len() + 1);
did.write_to_vec(&mut prefix);
prefix.push(REC_SEP);
prefix
}
// prefix format: {DID} 00 (DID trimmed)
pub fn record_prefix_did(did: &Did) -> Vec<u8> {
record_prefix_trimmed(&TrimmedDid::from(did))
}
// prefix format: {DID} 00 {collection} 2f
pub fn record_prefix_collection(did: &Did, collection: &str) -> Vec<u8> {
let mut prefix = record_prefix_did(did);
prefix.reserve(collection.len() + 1);
prefix.extend_from_slice(collection.as_bytes());
prefix.push(REC_PATH_SEP);
prefix
}
// key format: {DID} 00 {collection} 2f {rkey text}
//
// the rkey is stored as its canonical text (13-char base32 for TIDs), not
// tagged bytes: text order is MST order, so a repo's records lay out in
// exactly MST walk order and `generate_car` no longer re-sorts.
pub fn record_key_trimmed(did: &TrimmedDid, collection: &str, rkey: &DbRkey) -> Vec<u8> {
let rkey = rkey.to_smolstr();
let mut key = Vec::with_capacity(did.len() + 1 + collection.len() + 1 + rkey.len());
did.write_to_vec(&mut key);
key.push(REC_SEP);
key.extend_from_slice(collection.as_bytes());
key.push(REC_PATH_SEP);
key.extend_from_slice(rkey.as_bytes());
key
}
// key format: {DID} 00 {collection} 2f {rkey text}
pub fn record_key(did: &Did, collection: &str, rkey: &DbRkey) -> Vec<u8> {
record_key_trimmed(&TrimmedDid::from(did), collection, rkey)
}
/// parse an rkey from its text encoding. TID-shaped text becomes
/// [`DbRkey::Tid`], everything else [`DbRkey::Str`]; the text itself is the
/// lossless form, so the distinction only affects in-memory handling.
pub fn parse_rkey_text(raw: &[u8]) -> miette::Result<DbRkey> {
let s = std::str::from_utf8(raw)
.into_diagnostic()
.wrap_err("record key has invalid rkey utf8")?;
Ok(DbRkey::new(s))
}
/// split a record key's `{collection} 2f {rkey}` suffix (everything after the
/// `{DID} 00` prefix) into its parts.
pub fn split_record_suffix(suffix: &[u8]) -> miette::Result<(&str, DbRkey)> {
let sep = suffix
.iter()
.position(|&b| b == REC_PATH_SEP)
.ok_or_else(|| miette::miette!("record key missing collection/rkey separator"))?;
let collection = std::str::from_utf8(&suffix[..sep])
.into_diagnostic()
.wrap_err("record key has invalid collection utf8")?;
let rkey = parse_rkey_text(&suffix[sep + 1..])?;
Ok((collection, rkey))
}
// key format: {record key} 00 {death_rev: u64 BE}
//
// a record's superseded bodies, keyed by the rev at which they died. the NUL
// sorts a record's history immediately after the record itself, and BE revs
// sort deaths chronologically within it.
pub fn history_key(record_key: &[u8], death_rev: &DbTid) -> Vec<u8> {
let mut key = Vec::with_capacity(record_key.len() + 9);
key.extend_from_slice(record_key);
key.push(REC_SEP);
key.extend_from_slice(death_rev.as_bytes());
key
}
/// durable operator-redaction identity: `{record key} 00 {cid}`.
///
/// this lives in its own keyspace. keeping the cid in the key makes replay
/// suppression a point read rather than an unbounded scan of record history.
pub fn redaction_key(record_key: &[u8], cid: &jacquard_common::types::cid::IpldCid) -> Vec<u8> {
record_cid_key(record_key, cid)
}
fn record_cid_key(record_key: &[u8], cid: &jacquard_common::types::cid::IpldCid) -> Vec<u8> {
let cid = cid.to_bytes();
let mut key = Vec::with_capacity(record_key.len() + 1 + cid.len());
key.extend_from_slice(record_key);
key.push(REC_SEP);
key.extend_from_slice(&cid);
key
}
/// split a `{record key} 00 {cid}` key and require hydrant's canonical
/// CIDv1/dag-cbor/sha256 record identity.
pub fn split_record_cid_key(key: &[u8]) -> Option<(&[u8], jacquard_common::types::cid::IpldCid)> {
const CID_LEN: usize = 36;
let record_end = key.len().checked_sub(CID_LEN + 1)?;
if key.get(record_end) != Some(&REC_SEP) {
return None;
}
let raw = &key[record_end + 1..];
if !raw.starts_with(&[0x01, 0x71, 0x12, 0x20]) {
return None;
}
let cid = cid::Cid::read_bytes(raw).ok()?;
(cid.to_bytes().as_slice() == raw).then_some((&key[..record_end], cid))
}
/// exact body location for a pre-v10 pointer event.
///
/// the cid is part of the key because forks can produce two different bodies
/// for the same record and revision. unlike ordinary history, this finite
/// compatibility archive does not pretend an event's birth revision is a death
/// time.
#[cfg(feature = "indexer_stream")]
pub fn event_body_key(record_key: &[u8], cid: &jacquard_common::types::cid::IpldCid) -> Vec<u8> {
record_cid_key(record_key, cid)
}
/// the range of history entries for `record_key` with death rev strictly
/// greater than `after`: (`key` 00 `after`, `key` 01).
///
/// read-side only: stream inflation is today's sole history reader
///
/// the NUL/01 byte pair brackets this record's entries exactly in byte order:
/// a longer rkey sharing this one as a prefix continues with a byte above 01,
/// so it sorts past the upper bound. [`history_hit_for`] checks hits anyway.
#[cfg(feature = "indexer_stream")]
pub fn history_range_after(
record_key: &[u8],
after: &DbTid,
) -> (std::ops::Bound<Vec<u8>>, std::ops::Bound<Vec<u8>>) {
use std::ops::Bound;
let mut lower = Vec::with_capacity(record_key.len() + 9);
lower.extend_from_slice(record_key);
lower.push(REC_SEP);
lower.extend_from_slice(after.as_bytes());
let mut upper = Vec::with_capacity(record_key.len() + 1);
upper.extend_from_slice(record_key);
upper.push(REC_SEP + 1);
(Bound::Excluded(lower), Bound::Excluded(upper))
}
/// a history range hit only names this record if it is exactly
/// `{record key} 00 {rev}`; longer records sharing the prefix must be rejected.
#[cfg(feature = "indexer_stream")]
pub fn history_hit_for(record_key: &[u8], hit_key: &[u8]) -> bool {
hit_key.len() == record_key.len() + 9 && hit_key.starts_with(record_key)
}
// key format: r|{DID}|{collection} (DID trimmed)
pub fn count_collection_key(did: &Did, collection: &str) -> Vec<u8> {
let mut key = super::did_collection_prefix(did);
key.extend_from_slice(collection.as_bytes());
key
}
// key format: {DID}|{rev}
pub fn resync_buffer_key(did: &Did, rev: DbTid) -> Vec<u8> {
let repo = TrimmedDid::from(did);
let mut key = Vec::with_capacity(repo.len() + 1 + 8);
repo.write_to_vec(&mut key);
key.push(SEP);
key.extend_from_slice(rev.as_bytes());
key
}
// prefix format: {DID}| (DID trimmed)
pub fn resync_buffer_prefix(did: &Did) -> Vec<u8> {
let repo = TrimmedDid::from(did);
let mut prefix = Vec::with_capacity(repo.len() + 1);
repo.write_to_vec(&mut prefix);
prefix.push(SEP);
prefix
}
/// key format: `ret|<did bytes>`
pub const CRAWLER_RETRY_PREFIX: &[u8] = b"ret|";
pub fn crawler_retry_key(did: &Did) -> Vec<u8> {
let repo = TrimmedDid::from(did);
let mut key = Vec::with_capacity(CRAWLER_RETRY_PREFIX.len() + repo.len());
key.extend_from_slice(CRAWLER_RETRY_PREFIX);
repo.write_to_vec(&mut key);
key
}
pub fn crawler_retry_parse_key(key: &[u8]) -> miette::Result<TrimmedDid<'_>> {
let did = key
.strip_prefix(CRAWLER_RETRY_PREFIX)
.ok_or_else(|| miette::miette!("invalid crawler retry key"))?;
TrimmedDid::try_from(did)
}
/// key format: `job|<did bytes>`, a bulk job the repo's indexer shard is running for it
pub const BULK_JOB_PREFIX: &[u8] = b"job|";
pub fn bulk_job_key(did: &Did) -> Vec<u8> {
let repo = TrimmedDid::from(did);
let mut key = Vec::with_capacity(BULK_JOB_PREFIX.len() + repo.len());
key.extend_from_slice(BULK_JOB_PREFIX);
repo.write_to_vec(&mut key);
key
}
pub fn bulk_job_parse_key(key: &[u8]) -> miette::Result<TrimmedDid<'_>> {
let did = key
.strip_prefix(BULK_JOB_PREFIX)
.ok_or_else(|| miette::miette!("invalid bulk job key"))?;
TrimmedDid::try_from(did)
}
pub const CRAWLER_CURSOR_PREFIX: &[u8] = b"crawler_cursor|";
pub fn crawler_cursor_key(relay: &str) -> Vec<u8> {
let mut key = CRAWLER_CURSOR_PREFIX.to_vec();
key.extend_from_slice(relay.as_bytes());
key
}
pub const RELAY_FIRST_CURSOR_PREFIX: &[u8] = b"relay_first_cursor|";
pub fn relay_first_cursor_prefix(relay: &str) -> Vec<u8> {
let mut prefix = Vec::with_capacity(RELAY_FIRST_CURSOR_PREFIX.len() + relay.len() + 1);
prefix.extend_from_slice(RELAY_FIRST_CURSOR_PREFIX);
prefix.extend_from_slice(relay.as_bytes());
prefix.push(SEP);
prefix
}
fn relay_first_cursor_key(relay: &str, suffix: &str) -> Vec<u8> {
let mut key = relay_first_cursor_prefix(relay);
key.extend_from_slice(suffix.as_bytes());
key
}
pub fn relay_first_pass_key(relay: &str) -> Vec<u8> {
relay_first_cursor_key(relay, "pass")
}
pub fn relay_first_handled_key(relay: &str) -> Vec<u8> {
relay_first_cursor_key(relay, "handled")
}
pub fn relay_first_hosts_cursor_key(relay: &str) -> Vec<u8> {
relay_first_cursor_key(relay, "hosts")
}
pub fn relay_first_pds_cursor_key(relay: &str, host: &str) -> Vec<u8> {
relay_first_cursor_key(relay, &format!("pds|{host}"))
}
/// key format: `new_pds_crawl|<host>`, a PDS we started following that still
/// owes us one `listRepos` walk. the value is its page cursor.
pub const NEW_PDS_CRAWL_PREFIX: &[u8] = b"new_pds_crawl|";
pub fn new_pds_crawl_key(host: &str) -> Vec<u8> {
let mut key = Vec::with_capacity(NEW_PDS_CRAWL_PREFIX.len() + host.len());
key.extend_from_slice(NEW_PDS_CRAWL_PREFIX);
key.extend_from_slice(host.as_bytes());
key
}
/// key format: `rewalk|<host>`, a `listRepos` walk a followed PDS asked for
/// through `requestCrawl`. the value is its page cursor.
pub const REWALK_PREFIX: &[u8] = b"rewalk|";
pub fn rewalk_key(host: &str) -> Vec<u8> {
let mut key = Vec::with_capacity(REWALK_PREFIX.len() + host.len());
key.extend_from_slice(REWALK_PREFIX);
key.extend_from_slice(host.as_bytes());
key
}
/// key format: `refetch_allowance|<host>`, the `Allowance` all repos on that pds share.
pub fn refetch_allowance_key(host: &str) -> Vec<u8> {
let prefix = b"refetch_allowance|";
let mut key = Vec::with_capacity(prefix.len() + host.len());
key.extend_from_slice(prefix);
key.extend_from_slice(host.as_bytes());
key
}
/// key format: `rewalk_allowance|<host>`, the `listRepos` bytes a PDS's rewalks read in its current window.
pub fn rewalk_allowance_key(host: &str) -> Vec<u8> {
let prefix = b"rewalk_allowance|";
let mut key = Vec::with_capacity(prefix.len() + host.len());
key.extend_from_slice(prefix);
key.extend_from_slice(host.as_bytes());
key
}
/// key format: `seed_listed|<seed url>`, set once a seed relay's `listHosts`
/// has been read to the end with every host on it admitted.
pub const SEED_LISTED_PREFIX: &[u8] = b"seed_listed|";
pub fn seed_listed_key(seed: &str) -> Vec<u8> {
let mut key = Vec::with_capacity(SEED_LISTED_PREFIX.len() + seed.len());
key.extend_from_slice(SEED_LISTED_PREFIX);
key.extend_from_slice(seed.as_bytes());
key
}
pub const BY_COLLECTION_CURSOR_PREFIX: &[u8] = b"by_collection_cursor|";
/// prefix for all by-collection cursors belonging to a given index URL.
pub fn by_collection_cursor_prefix(url: &str) -> Vec<u8> {
let mut prefix = BY_COLLECTION_CURSOR_PREFIX.to_vec();
prefix.extend_from_slice(url.as_bytes());
prefix.push(SEP);
prefix
}
pub fn by_collection_cursor_key(url: &str, collection: &str) -> Vec<u8> {
let mut key = by_collection_cursor_prefix(url);
key.extend_from_slice(collection.as_bytes());
key
}
pub const CRAWLER_SOURCE_PREFIX: &[u8] = b"src|";
pub fn crawler_source_key(url: &str) -> Vec<u8> {
let mut key = Vec::with_capacity(CRAWLER_SOURCE_PREFIX.len() + url.len());
key.extend_from_slice(CRAWLER_SOURCE_PREFIX);
key.extend_from_slice(url.as_bytes());
key
}
// key format: {collection}|{cid_bytes}
pub fn block_key(collection: &str, cid: &[u8]) -> Vec<u8> {
let mut key = Vec::with_capacity(collection.len() + 1 + cid.len());
key.extend_from_slice(collection.as_bytes());
key.push(SEP);
key.extend_from_slice(cid);
key
}
#[cfg(test)]
mod tests {
use super::*;
use jacquard_common::types::string::Tid;
fn did() -> Did {
Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap()
}
fn tid_rkey(s: &str) -> DbRkey {
DbRkey::Tid(DbTid::from(&Tid::new(s).unwrap()))
}
#[test]
fn record_key_roundtrips_collection_and_rkey() {
let did = did();
for rkey in [
DbRkey::Str("self".into()),
DbRkey::Str("abc".into()),
DbRkey::Str("~tilde".into()),
tid_rkey("3kzbif5moe22m"),
] {
let key = record_key(&did, "app.bsky.feed.post", &rkey);
let prefix = record_prefix_did(&did);
assert!(key.starts_with(&prefix));
let (col, parsed) = split_record_suffix(&key[prefix.len()..]).unwrap();
assert_eq!(col, "app.bsky.feed.post");
assert_eq!(parsed.to_smolstr(), rkey.to_smolstr());
}
}
#[test]
fn record_keys_sort_in_mst_order() {
let did = did();
let mut keys: Vec<Vec<u8>> = [
("app.bsky.actor.profile", DbRkey::Str("self".into())),
("app.bsky.feed.like", tid_rkey("3kzbif5moe22m")),
("app.bsky.feed.like", DbRkey::Str("abc".into())),
("app.bsky.feed.like", DbRkey::Str("abcd".into())),
("app.bsky.feed.post", tid_rkey("3kzbif5abc22a")),
("app.bsky.feed.post", DbRkey::Str("z-string".into())),
]
.into_iter()
.map(|(col, rk)| record_key(&did, col, &rk))
.collect();
let mut sorted = keys.clone();
sorted.sort();
// mst order is plain lexicographic order of `{collection}/{rkey}` text
keys.sort_by_key(|k| {
let prefix_len = record_prefix_did(&did).len();
k[prefix_len..].to_vec()
});
assert_eq!(sorted, keys);
// and the prefix pair nests correctly: "abc" before "abcd"
let a = record_key(&did, "app.bsky.feed.like", &DbRkey::Str("abc".into()));
let b = record_key(&did, "app.bsky.feed.like", &DbRkey::Str("abcd".into()));
assert!(a < b);
}
#[test]
fn new_record_keys_sort_before_legacy_keys() {
// the v10 migration scans ascending and must only meet legacy keys:
// new keys (`did 00`) sort before legacy keys (`did 7c`) of the same repo
let did = did();
let new = record_key(&did, "app.bsky.feed.post", &DbRkey::Str("self".into()));
let mut legacy = record_prefix_did(&did);
*legacy.last_mut().unwrap() = super::SEP;
legacy.extend_from_slice(b"app.bsky.feed.post|sself");
assert!(new < legacy);
}
#[cfg(feature = "indexer_stream")]
#[test]
fn history_key_brackets_and_hit_guard() {
let did = did();
let rec = record_key(&did, "app.bsky.feed.post", &DbRkey::Str("abc".into()));
let longer = record_key(&did, "app.bsky.feed.post", &DbRkey::Str("abcd".into()));
let rev = DbTid::from(&Tid::new("3kzbif5moe22m").unwrap());
let death = history_key(&rec, &rev);
// a record's history sorts immediately after the record itself
assert!(rec < death);
// the range finds this record's deaths, including one at rev+1
let later = DbTid::from(&Tid::new("3kzbif5moe22n").unwrap());
let later_death = history_key(&rec, &later);
let (lo, hi) = history_range_after(&rec, &rev);
let in_range = |k: &Vec<u8>| {
(match &lo {
std::ops::Bound::Excluded(l) => k > l,
_ => unreachable!(),
}) && (match &hi {
std::ops::Bound::Excluded(h) => k < h,
_ => unreachable!(),
})
};
assert!(in_range(&later_death));
// a death exactly at `after` is excluded (strictly-after semantics)
assert!(!in_range(&death));
// a longer record sharing this one as a prefix sorts above the upper
// bound (`rec 01` < `rec d`), so the bounded range only ever contains
// this record's own `rec 00 rev` entries...
let longer_death = history_key(&longer, &later);
assert!(!in_range(&longer));
assert!(!in_range(&longer_death));
// ...and a different record entirely also sorts outside
let other = record_key(&did, "app.bsky.feed.post", &DbRkey::Str("abd".into()));
assert!(!in_range(&other));
// the guard is defense in depth: even if a foreign key were hit (e.g.
// via an unbounded seek), it is accepted only for this exact record
assert!(history_hit_for(&rec, &death));
assert!(!history_hit_for(&rec, &longer));
assert!(!history_hit_for(&rec, &longer_death));
}
#[test]
fn history_death_revs_sort_chronologically() {
let did = did();
let rec = record_key(&did, "app.bsky.feed.post", &tid_rkey("3kzbif5moe22m"));
let early = history_key(&rec, &DbTid::from(&Tid::new("3kzbif5moe22m").unwrap()));
let late = history_key(&rec, &DbTid::from(&Tid::new("3kzbif5mof33m").unwrap()));
assert!(early < late);
}
#[test]
fn indexer_key_codec_inventory_contracts() {
let did = did();
let trimmed = TrimmedDid::from(&did);
let rkey = DbRkey::Str("stinkpot".into());
let rev = DbTid::from(&Tid::new("3kzbif5moe22m").unwrap());
assert_eq!(pending_key(42), 42_u64.to_be_bytes());
let repo_key = crate::db::keys::repo_key(&did);
assert_eq!(
pending_did_key(&repo_key, &pending_key(42)),
[repo_key.as_slice(), &[REC_SEP], &pending_key(42)].concat()
);
// a did that prefixes another sorts first here too, the same as in `resync`
let short = crate::db::keys::repo_key(&Did::new_static("did:web:a.com").unwrap());
let long = crate::db::keys::repo_key(&Did::new_static("did:web:a.com.au").unwrap());
assert!(short < long);
assert!(
pending_did_key(&short, &pending_key(u64::MAX))
< pending_did_key(&long, &pending_key(0))
);
#[cfg(feature = "indexer_stream")]
assert_eq!(
event_watermark_key(42),
[EVENT_WATERMARK_PREFIX, 42_u64.to_be_bytes().as_slice()].concat()
);
let record_prefix = record_prefix_trimmed(&trimmed);
assert_eq!(trimmed_did_len(&record_prefix), Some(trimmed.len()));
assert_eq!(record_prefix, record_prefix_did(&did));
let collection_prefix = record_prefix_collection(&did, "app.bsky.feed.post");
let record = record_key_trimmed(&trimmed, "app.bsky.feed.post", &rkey);
assert_eq!(record, record_key(&did, "app.bsky.feed.post", &rkey));
assert!(record.starts_with(&collection_prefix));
assert_eq!(parse_rkey_text(b"stinkpot").unwrap(), rkey);
assert!(parse_rkey_text(&[0xff]).is_err());
assert!(split_record_suffix(b"missing-separator").is_err());
let cid = crate::car::compute_cid(b"body");
let redaction = redaction_key(&record, &cid);
assert_eq!(
split_record_cid_key(&redaction),
Some((record.as_slice(), cid))
);
assert!(split_record_cid_key(&record).is_none());
#[cfg(feature = "indexer_stream")]
assert_eq!(event_body_key(&record, &cid), redaction);
let count_prefix = super::super::did_collection_prefix(&did);
assert_eq!(
count_collection_key(&did, "app.bsky.feed.post"),
[count_prefix.as_slice(), b"app.bsky.feed.post"].concat()
);
let resync_prefix = resync_buffer_prefix(&did);
assert_eq!(
resync_buffer_key(&did, rev),
[resync_prefix.as_slice(), rev.as_bytes()].concat()
);
let retry = crawler_retry_key(&did);
assert_eq!(crawler_retry_parse_key(&retry).unwrap(), trimmed);
assert!(crawler_retry_parse_key(b"bad").is_err());
assert!(crawler_retry_parse_key(b"nope|did:web:example.com").is_err());
let job = bulk_job_key(&did);
assert!(job.starts_with(BULK_JOB_PREFIX));
assert_eq!(bulk_job_parse_key(&job).unwrap(), trimmed);
assert!(bulk_job_parse_key(&retry).is_err());
assert_eq!(
crawler_cursor_key("https://relay.example"),
b"crawler_cursor|https://relay.example"
);
assert_eq!(
relay_first_cursor_prefix("https://relay.example"),
b"relay_first_cursor|https://relay.example|"
);
assert_eq!(
relay_first_pass_key("https://relay.example"),
b"relay_first_cursor|https://relay.example|pass"
);
assert_eq!(
relay_first_handled_key("https://relay.example"),
b"relay_first_cursor|https://relay.example|handled"
);
assert_eq!(
relay_first_hosts_cursor_key("https://relay.example"),
b"relay_first_cursor|https://relay.example|hosts"
);
assert_eq!(
relay_first_pds_cursor_key("https://relay.example", "pds.example"),
b"relay_first_cursor|https://relay.example|pds|pds.example"
);
assert_eq!(
by_collection_cursor_prefix("https://relay.example"),
b"by_collection_cursor|https://relay.example|"
);
assert_eq!(
by_collection_cursor_key("https://relay.example", "app.bsky.feed.post"),
b"by_collection_cursor|https://relay.example|app.bsky.feed.post"
);
assert_eq!(
crawler_source_key("https://relay.example"),
b"src|https://relay.example"
);
let new_pds = new_pds_crawl_key("pds.example");
assert_eq!(new_pds, b"new_pds_crawl|pds.example");
assert_eq!(&new_pds[NEW_PDS_CRAWL_PREFIX.len()..], b"pds.example");
let rewalk = rewalk_key("pds.example");
assert_eq!(rewalk, b"rewalk|pds.example");
assert_eq!(&rewalk[REWALK_PREFIX.len()..], b"pds.example");
assert_eq!(
refetch_allowance_key("pds.example"),
b"refetch_allowance|pds.example"
);
assert_eq!(
rewalk_allowance_key("pds.example"),
b"rewalk_allowance|pds.example"
);
let seed = seed_listed_key("https://relay.example/");
assert_eq!(seed, b"seed_listed|https://relay.example/");
assert!(seed.starts_with(SEED_LISTED_PREFIX));
assert_eq!(
block_key("app.bsky.feed.post", &[1, 2, 3]),
b"app.bsky.feed.post|\x01\x02\x03"
);
}
}