use jacquard_common::types::string::Did;
use crate::db::types::TrimmedDid;
pub mod v1;
pub use v1::{firehose_cursor_key, firehose_cursor_key_from_url};
/// separator used for composite keys
pub const SEP: u8 = b'|';
#[cfg(feature = "indexer")]
pub mod indexer;
#[cfg(feature = "indexer")]
pub use indexer::*;
#[cfg(feature = "relay")]
pub const RELAY_EVENT_WATERMARK_PREFIX: &[u8] = b"rwm|";
/// THIS SHOULD ALWAYS BE STABLE. DO NOT CHANGE
pub const VERSIONING_KEY: &[u8] = b"db_version";
// key format: {DID}
pub fn repo_key(did: &Did) -> Vec<u8> {
let mut vec = Vec::with_capacity(32);
TrimmedDid::from(did).write_to_vec(&mut vec);
vec
}
pub const REPO_METADATA_PREFIX: &[u8] = b"rm|";
pub fn repo_metadata_key(did: &Did) -> Vec<u8> {
let mut vec = Vec::with_capacity(REPO_METADATA_PREFIX.len() + 32);
vec.extend_from_slice(REPO_METADATA_PREFIX);
TrimmedDid::from(did).write_to_vec(&mut vec);
vec
}
#[cfg(feature = "relay")]
/// key format: {SEQ} (u64 big-endian), mirroring event_key
pub fn relay_event_key(seq: u64) -> [u8; 8] {
seq.to_be_bytes()
}
#[cfg(feature = "relay")]
pub fn relay_event_watermark_key(timestamp_secs: u64) -> Vec<u8> {
let mut key = Vec::with_capacity(RELAY_EVENT_WATERMARK_PREFIX.len() + 8);
key.extend_from_slice(RELAY_EVENT_WATERMARK_PREFIX);
key.extend_from_slice(×tamp_secs.to_be_bytes());
key
}
// key format: {SEQ}
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
pub fn event_key(seq: u64) -> [u8; 8] {
seq.to_be_bytes()
}
#[cfg(feature = "jetstream")]
pub fn jetstream_event_key(time_us: u64, id: u64) -> [u8; 16] {
let mut key = [0u8; 16];
key[..8].copy_from_slice(&time_us.to_be_bytes());
key[8..].copy_from_slice(&id.to_be_bytes());
key
}
#[cfg(feature = "jetstream")]
pub fn parse_jetstream_event_key(key: &[u8]) -> miette::Result<(u64, u64)> {
if key.len() != 16 {
miette::bail!("jetstream event key is not 16 bytes");
}
let time_us = u64::from_be_bytes(
key[..8]
.try_into()
.map_err(|e| miette::miette!("invalid jetstream time bytes: {e}"))?,
);
let id = u64::from_be_bytes(
key[8..]
.try_into()
.map_err(|e| miette::miette!("invalid jetstream id bytes: {e}"))?,
);
Ok((time_us, id))
}
pub const COUNT_KS_PREFIX: &[u8] = &[b'k', SEP];
// count keys for the counts keyspace
// key format: k\x00{keyspace_name}
pub fn count_keyspace_key(name: &str) -> Vec<u8> {
let mut key = Vec::with_capacity(COUNT_KS_PREFIX.len() + name.len());
key.extend_from_slice(COUNT_KS_PREFIX);
key.extend_from_slice(name.as_bytes());
key
}
pub const COUNT_DELTA_PREFIX: &[u8] = &[b'd', SEP];
pub fn count_delta_key(id: u64, name: &str) -> Vec<u8> {
let mut key = Vec::with_capacity(COUNT_DELTA_PREFIX.len() + 8 + 1 + name.len());
key.extend_from_slice(COUNT_DELTA_PREFIX);
key.extend_from_slice(&id.to_be_bytes());
key.push(SEP);
key.extend_from_slice(name.as_bytes());
key
}
pub fn count_delta_start_key(id: u64) -> Vec<u8> {
let mut key = Vec::with_capacity(COUNT_DELTA_PREFIX.len() + 8);
key.extend_from_slice(COUNT_DELTA_PREFIX);
key.extend_from_slice(&id.to_be_bytes());
key
}
pub fn parse_count_delta_key(key: &[u8]) -> miette::Result<(u64, &str)> {
let min_len = COUNT_DELTA_PREFIX.len() + 8 + 1;
if key.len() < min_len || !key.starts_with(COUNT_DELTA_PREFIX) {
miette::bail!("invalid count delta key");
}
let id_start = COUNT_DELTA_PREFIX.len();
let id_end = id_start + 8;
let id = u64::from_be_bytes(
key[id_start..id_end]
.try_into()
.map_err(|e| miette::miette!("invalid count delta key id: {e}"))?,
);
if key[id_end] != SEP {
miette::bail!("invalid count delta key separator");
}
let name = std::str::from_utf8(&key[id_end + 1..])
.map_err(|e| miette::miette!("invalid count delta key name: {e}"))?;
Ok((id, name))
}
pub const COUNT_DELTA_WATERMARK_KEY: &[u8] = b"w|count_delta_watermark";
#[cfg(feature = "indexer")]
pub const COUNT_GAUGE_PREFIX: &[u8] = &[b'g', SEP];
#[cfg(feature = "indexer")]
pub fn count_gauge_key(did: &Did) -> Vec<u8> {
let repo = TrimmedDid::from(did);
let mut key = Vec::with_capacity(COUNT_GAUGE_PREFIX.len() + repo.len());
key.extend_from_slice(COUNT_GAUGE_PREFIX);
repo.write_to_vec(&mut key);
key
}
pub const COUNT_COLLECTION_PREFIX: &[u8] = &[b'r', SEP];
pub fn did_collection_prefix(did: &Did) -> Vec<u8> {
let repo = TrimmedDid::from(did);
let mut key = Vec::with_capacity(COUNT_COLLECTION_PREFIX.len() + repo.len() + 1);
key.extend_from_slice(COUNT_COLLECTION_PREFIX);
repo.write_to_vec(&mut key);
key.push(SEP);
key
}
pub const FIREHOSE_CURSOR_PREFIX: &[u8] = b"firehose_cursor|";
// stream position marks: `stream_next|{counter}` -> a reservation (u64 BE), a
// position no commit of that counter has reached, moved a block ahead by the
// commit that runs past it
#[cfg(feature = "indexer_stream")]
pub const STREAM_EVENT_IDS_MARK: &[u8] = b"stream_next|event_ids";
#[cfg(feature = "indexer_stream")]
pub const STREAM_ID: &[u8] = b"stream|id";
/// written in the same batch as the body change so a crash can't leave it unmarked
#[cfg(feature = "indexer_stream")]
pub const STREAM_CHUNKS_DIRTY_PREFIX: &[u8] = b"stream_chunks|dirty|";
#[cfg(feature = "jetstream")]
pub const JETSTREAM_IDS_MARK: &[u8] = b"stream_next|jetstream_ids";
#[cfg(feature = "jetstream")]
pub const JETSTREAM_TIME_US_MARK: &[u8] = b"stream_next|jetstream_time_us";
#[cfg(feature = "relay")]
pub const RELAY_SEQS_MARK: &[u8] = b"stream_next|relay_seqs";
pub const FIREHOSE_SOURCE_PREFIX: &[u8] = b"firehose|";
pub fn firehose_source_key(url: &str) -> Vec<u8> {
let mut key = Vec::with_capacity(FIREHOSE_SOURCE_PREFIX.len() + url.len());
key.extend_from_slice(FIREHOSE_SOURCE_PREFIX);
key.extend_from_slice(url.as_bytes());
key
}
/// durable marker for explicitly trusted operator/admin sources. Marker absence is public.
pub const TRUSTED_FIREHOSE_SOURCE_PREFIX: &[u8] = b"trusted_firehose_source/";
pub fn trusted_firehose_source_key(url: &str) -> Vec<u8> {
let mut key = Vec::with_capacity(TRUSTED_FIREHOSE_SOURCE_PREFIX.len() + url.len());
key.extend_from_slice(TRUSTED_FIREHOSE_SOURCE_PREFIX);
key.extend_from_slice(url.as_bytes());
key
}
pub fn pds_account_count_key(host: &str) -> String {
format!("p|{host}")
}
/// key for the persisted daily-PDS-add counter in the cursors keyspace.
/// value layout: [day: u64 BE][count: u64 BE] = 16 bytes.
pub const PDS_DAILY_ADDS_KEY: &[u8] = b"pds_daily_adds";
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn shared_key_codec_inventory_contracts() {
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
let repo = repo_key(&did);
assert_eq!(
repo_metadata_key(&did),
[REPO_METADATA_PREFIX, repo.as_slice()].concat()
);
assert_eq!(count_keyspace_key("heads"), b"k|heads");
let delta = count_delta_key(42, "heads");
assert_eq!(parse_count_delta_key(&delta).unwrap(), (42, "heads"));
assert_eq!(
count_delta_start_key(42),
&delta[..COUNT_DELTA_PREFIX.len() + 8]
);
assert!(parse_count_delta_key(b"bad").is_err());
let mut bad_separator = delta.clone();
bad_separator[COUNT_DELTA_PREFIX.len() + 8] = b'/';
assert!(parse_count_delta_key(&bad_separator).is_err());
let mut bad_utf8 = count_delta_start_key(42);
bad_utf8.extend_from_slice(&[SEP, 0xff]);
assert!(parse_count_delta_key(&bad_utf8).is_err());
#[cfg(feature = "indexer")]
assert_eq!(
count_gauge_key(&did),
[COUNT_GAUGE_PREFIX, repo.as_slice()].concat()
);
assert_eq!(
did_collection_prefix(&did),
[COUNT_COLLECTION_PREFIX, repo.as_slice(), &[SEP]].concat()
);
assert_eq!(
firehose_source_key("https://relay.example"),
b"firehose|https://relay.example"
);
assert_eq!(
trusted_firehose_source_key("wss://pds.example"),
b"trusted_firehose_source/wss://pds.example"
);
assert_eq!(pds_account_count_key("pds.example"), "p|pds.example");
assert_eq!(
firehose_cursor_key("relay.example"),
b"firehose_cursor|relay.example"
);
let url = url::Url::parse("https://relay.example:8443/xrpc").unwrap();
assert_eq!(
firehose_cursor_key_from_url(&url),
b"firehose_cursor|relay.example:8443"
);
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
assert_eq!(event_key(42), 42_u64.to_be_bytes());
#[cfg(feature = "relay")]
{
assert_eq!(relay_event_key(42), 42_u64.to_be_bytes());
assert_eq!(
relay_event_watermark_key(42),
[
RELAY_EVENT_WATERMARK_PREFIX,
42_u64.to_be_bytes().as_slice()
]
.concat()
);
}
#[cfg(feature = "jetstream")]
{
let key = jetstream_event_key(42, 7);
assert_eq!(parse_jetstream_event_key(&key).unwrap(), (42, 7));
assert!(parse_jetstream_event_key(&key[..15]).is_err());
}
}
}