use crate::api::AppState;
#[cfg(feature = "indexer")]
use crate::db::keys;
use crate::db::registry;
#[cfg(feature = "indexer_stream")]
use crate::db::types::{DbAction, DbRkey, DbTid, DidKey, TrimmedDid};
#[cfg(feature = "indexer_stream")]
use crate::types::{Commit, RepoMetadata, RepoState, RepoStatus, StoredData, StoredEvent};
use axum::routing::{get, post};
use axum::{
Json,
extract::{Query, State},
http::StatusCode,
};
#[cfg(feature = "indexer_stream")]
use bytes::Bytes;
#[cfg(feature = "indexer_stream")]
use jacquard_common::CowStr;
#[cfg(feature = "indexer")]
use jacquard_common::types::ident::AtIdentifier;
#[cfg(feature = "indexer_stream")]
use jacquard_common::types::{did::Did, tid::Tid};
use miette::IntoDiagnostic;
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[cfg(feature = "indexer_stream")]
use std::borrow::Cow;
use std::sync::Arc;
#[cfg(feature = "indexer")]
#[derive(Deserialize)]
pub struct DebugCountRequest {
pub did: String,
pub collection: String,
}
#[cfg(feature = "indexer")]
#[derive(Serialize)]
pub struct DebugCountResponse {
pub count: usize,
}
#[cfg(feature = "indexer_stream")]
#[derive(Serialize)]
pub struct DebugSeedXrpcRepoResponse {
pub did: String,
pub collection: &'static str,
pub rkey: &'static str,
pub cid: String,
pub commit_cid: String,
pub rev: String,
pub event_id: u64,
pub hostname: &'static str,
}
pub fn router() -> axum::Router<Arc<AppState>> {
let r = axum::Router::new()
.route("/debug/get", get(handle_debug_get))
.route("/debug/iter", get(handle_debug_iter))
.route("/debug/compact", post(handle_debug_compact));
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
let r = r
.route(
"/debug/ephemeral_ttl_tick",
post(handle_debug_ephemeral_ttl_tick),
)
.route("/debug/seed_watermark", post(handle_debug_seed_watermark))
.route("/debug/seed_events", post(handle_debug_seed_events));
#[cfg(all(feature = "relay", feature = "jetstream"))]
let r = r.route(
"/debug/seed_relay_jetstream_commits",
post(handle_debug_seed_relay_jetstream_commits),
);
#[cfg(all(feature = "indexer_stream", feature = "backlinks"))]
let r = r.route("/debug/seed_backlinks", post(handle_debug_seed_backlinks));
#[cfg(feature = "indexer")]
let r = r.route("/debug/count", get(handle_debug_count));
#[cfg(feature = "indexer_stream")]
let r = r.route("/debug/seed_xrpc_repo", post(handle_debug_seed_xrpc_repo));
r
}
#[cfg(feature = "indexer")]
pub async fn handle_debug_count(
State(state): State<Arc<AppState>>,
Query(req): Query<DebugCountRequest>,
) -> Result<Json<DebugCountResponse>, StatusCode> {
let ident = req
.did
.parse::<AtIdentifier>()
.map_err(|_| StatusCode::BAD_REQUEST)?;
let did = state
.resolver
.resolve_did(&ident)
.await
.map_err(|_| StatusCode::BAD_REQUEST)?;
let prefix = keys::record_prefix_collection(&did, &req.collection);
let count = state
.db
.run(move |db| {
let start_key = prefix.clone();
// rkeys cannot contain 0xff, so every key in the collection sorts
// below `prefix ff`
let mut end_key = prefix.clone();
end_key.push(0xff);
Ok::<_, miette::Report>(db.indexer.head_range(start_key..end_key).count())
})
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(Json(DebugCountResponse { count }))
}
#[cfg(feature = "indexer_stream")]
pub async fn handle_debug_seed_xrpc_repo(
State(state): State<Arc<AppState>>,
) -> Result<Json<DebugSeedXrpcRepoResponse>, StatusCode> {
const DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz";
const COLLECTION: &str = "app.bsky.feed.post";
const RKEY: &str = "stinkpot";
const REV: &str = "3jzfcijpj2z2a";
const HOSTNAME: &str = "pds.example";
let did: Did = Did::new_static(DID).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let rev = Tid::new(REV).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let body = Bytes::from(
serde_ipld_dagcbor::to_vec(&serde_json::json!({
"$type": COLLECTION,
"text": "eight-byte rkey fixture",
}))
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
);
let cid = crate::car::compute_cid(&body);
let (root, _) = crate::mst::build_tree([(format!("{COLLECTION}/{RKEY}"), cid)])
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let mut repo_state = RepoState::backfilling();
repo_state.status = RepoStatus::Synced;
repo_state.root = Some(Commit {
data: root,
prev: None,
rev: DbTid::from(&rev),
sig: Bytes::from_static(b"fixture-signature"),
version: 3,
});
repo_state.pds = Some(CowStr::Owned(format!("https://{HOSTNAME}").into()));
repo_state.signing_key = Some(DidKey(Cow::Owned(
[0xed, 0x01]
.into_iter()
.chain(std::iter::repeat_n(7, 32))
.collect(),
)));
let commit_cid = repo_state
.root
.clone()
.and_then(|commit| commit.into_atp_commit(did.clone()))
.ok_or(StatusCode::INTERNAL_SERVER_ERROR)?
.to_cid()
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let record_key = keys::record_key(&did, COLLECTION, &crate::db::types::DbRkey::new(RKEY));
let repo_key = keys::repo_key(&did);
let metadata_key = keys::repo_metadata_key(&did);
let collection_count_key = keys::count_collection_key(&did, COLLECTION);
let host_count_key = keys::pds_account_count_key(HOSTNAME);
let host_cursor_key = keys::firehose_cursor_key(HOSTNAME);
let stored_event = StoredEvent {
live: true,
did: TrimmedDid::from(&did).into_static(),
rev: DbTid::from(&rev),
collection: CowStr::Borrowed(COLLECTION),
rkey: DbRkey::new(RKEY),
action: DbAction::Create,
data: StoredData::Ptr(cid),
};
let event_id = state
.db
.run(move |db| {
let mut txn = crate::db::Txn::new(db);
db.indexer
.stage_head(&mut txn.batch, record_key, body.as_ref());
txn.batch
.insert(&db.repos, repo_key, crate::db::ser_repo_state(&repo_state)?);
txn.batch.insert(
&db.repo_metadata,
metadata_key,
crate::db::ser_repo_meta(&RepoMetadata::backfilling(1))?,
);
txn.batch
.insert(&db.counts, collection_count_key, 1_u64.to_be_bytes());
txn.batch
.insert(&db.counts, host_count_key, 1_u64.to_be_bytes());
txn.batch
.insert(&db.cursors, host_cursor_key, 42_i64.to_be_bytes());
let event = txn
.outbox
.stream
.push_record(rmp_serde::to_vec(&stored_event).into_diagnostic()?, None);
txn.commit_assigned()?
.stream_id(event)
.ok_or_else(|| miette::miette!("seeded event was not given an id"))
})
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let host_count_key = keys::pds_account_count_key(HOSTNAME);
let current_host_count = state.db.get_count_sync(&host_count_key);
state
.db
.update_count(&host_count_key, 1_i64 - current_host_count as i64);
crate::pds_meta::PdsMeta::update_host(&state.pds_meta, HOSTNAME, |host| {
host.status = crate::pds_meta::HostStatus::Active;
});
Ok(Json(DebugSeedXrpcRepoResponse {
did: DID.to_string(),
collection: COLLECTION,
rkey: RKEY,
cid: cid.to_string(),
commit_cid: commit_cid.to_string(),
rev: REV.to_string(),
event_id,
hostname: HOSTNAME,
}))
}
#[derive(Deserialize)]
pub struct DebugGetRequest {
pub partition: String,
pub key: String,
}
#[derive(Serialize)]
pub struct DebugGetResponse {
pub value: Option<Value>,
}
pub async fn handle_debug_get(
State(state): State<Arc<AppState>>,
Query(req): Query<DebugGetRequest>,
) -> Result<Json<DebugGetResponse>, StatusCode> {
let ks = get_keyspace_by_name(&state.db, &req.partition)?;
let key = registry::debug_parse_key(&req.partition, &req.key).ok_or(StatusCode::BAD_REQUEST)?;
let partition = req.partition.clone();
let value = state
.db
.run(move |db| {
ks.get(key)
.inspect_err(|e| db.poison.check(e))
.into_diagnostic()
})
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.map(|v| registry::debug_value(&partition, &v));
Ok(Json(DebugGetResponse { value }))
}
#[derive(Deserialize)]
pub struct DebugIterRequest {
pub partition: String,
pub start: Option<String>,
pub end: Option<String>,
pub limit: Option<usize>,
pub reverse: Option<bool>,
}
#[derive(Serialize)]
pub struct DebugIterResponse {
pub items: Vec<(String, Value)>,
}
pub async fn handle_debug_iter(
State(state): State<Arc<AppState>>,
Query(req): Query<DebugIterRequest>,
) -> Result<Json<DebugIterResponse>, StatusCode> {
let partition = req.partition.clone();
let parse_bound = |s: Option<String>| -> Result<Option<Vec<u8>>, StatusCode> {
s.map(|s| registry::debug_parse_key(&partition, &s).ok_or(StatusCode::BAD_REQUEST))
.transpose()
};
let start = parse_bound(req.start)?;
let end = parse_bound(req.end)?;
let items = state
.db
.run(move |db| {
let ks = get_keyspace_by_name(db, &req.partition)
.map_err(|_| miette::miette!("bad request"))?;
let limit = req.limit.unwrap_or(50);
let collect = |iter: &mut dyn Iterator<Item = fjall::Guard>| {
let mut items = Vec::new();
for guard in iter.take(limit) {
let (k, v) = guard
.into_inner()
.map_err(|_| miette::miette!("internal error"))?;
let key_str = registry::debug_render_key(&partition, &k);
items.push((key_str, registry::debug_value(&partition, &v)));
}
Ok::<_, miette::Report>(items)
};
let start_bound = if let Some(s) = &start {
std::ops::Bound::Included(s.as_slice())
} else {
std::ops::Bound::Unbounded
};
let end_bound = if let Some(e) = &end {
std::ops::Bound::Included(e.as_slice())
} else {
std::ops::Bound::Unbounded
};
if req.reverse == Some(true) {
collect(
&mut ks
.range::<&[u8], (std::ops::Bound<&[u8]>, std::ops::Bound<&[u8]>)>((
start_bound,
end_bound,
))
.rev(),
)
} else {
collect(
&mut ks.range::<&[u8], (std::ops::Bound<&[u8]>, std::ops::Bound<&[u8]>)>((
start_bound,
end_bound,
)),
)
}
})
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(Json(DebugIterResponse { items }))
}
fn get_keyspace_by_name(db: &crate::db::Db, name: &str) -> Result<fjall::Keyspace, StatusCode> {
db.keyspace_by_name(name).ok_or(StatusCode::BAD_REQUEST)
}
#[derive(Deserialize)]
pub struct DebugCompactRequest {
pub partition: String,
}
pub async fn handle_debug_compact(
State(state): State<Arc<AppState>>,
Query(req): Query<DebugCompactRequest>,
) -> Result<StatusCode, StatusCode> {
state
.db
.run(move |db| {
let ks = get_keyspace_by_name(db, &req.partition)
.map_err(|_| miette::miette!("bad request"))?;
ks.remove(b"dummy_tombstone123").into_diagnostic()?;
db.inner
.persist(fjall::PersistMode::Buffer)
.into_diagnostic()?;
ks.rotate_memtable_and_wait().into_diagnostic()?;
ks.major_compact().into_diagnostic()?;
Ok::<_, miette::Report>(())
})
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(StatusCode::OK)
}
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
pub async fn handle_debug_ephemeral_ttl_tick(
State(state): State<Arc<AppState>>,
) -> Result<StatusCode, StatusCode> {
let ephemeral_ttl = state.ephemeral_ttl;
state
.db
.run(move |db| {
#[cfg(feature = "indexer_stream")]
crate::db::ephemeral::ephemeral_ttl_tick(db, &ephemeral_ttl)?;
#[cfg(feature = "relay")]
crate::db::ephemeral::relay_events_ttl_tick(db, &ephemeral_ttl)?;
#[cfg(feature = "jetstream")]
crate::db::ephemeral::jetstream_events_ttl_tick(db, &ephemeral_ttl)?;
Ok::<_, miette::Report>(())
})
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(StatusCode::OK)
}
#[derive(Deserialize)]
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
pub struct DebugSeedWatermarkRequest {
/// unix timestamp (seconds) to write the watermark at
pub ts: u64,
/// event_id the watermark points to, all events before this id will be pruned
pub event_id: u64,
}
/// writes an event watermark entry directly to the cursors keyspace, using identical
/// key/value encoding to the real TTL worker. used in tests to plant a past watermark
/// so the real `ephemeral_ttl_tick` code path is exercised without waiting 3600 seconds.
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
pub async fn handle_debug_seed_watermark(
State(state): State<Arc<AppState>>,
Query(req): Query<DebugSeedWatermarkRequest>,
) -> Result<StatusCode, StatusCode> {
state
.db
.run(move |db| {
#[cfg(feature = "indexer_stream")]
db.cursors
.insert(
crate::db::keys::event_watermark_key(req.ts),
req.event_id.to_be_bytes(),
)
.into_diagnostic()?;
#[cfg(feature = "relay")]
db.cursors
.insert(
crate::db::keys::relay_event_watermark_key(req.ts),
req.event_id.to_be_bytes(),
)
.into_diagnostic()?;
Ok::<_, miette::Report>(())
})
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(StatusCode::OK)
}
/// the most events one seed request writes. a seed is one sequenced commit,
/// which holds up every other writer until it lands.
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
const MAX_SEEDED_EVENTS: usize = 10_000;
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
fn seed_count(count: u64) -> Result<usize, StatusCode> {
usize::try_from(count)
.ok()
.filter(|count| *count <= MAX_SEEDED_EVENTS)
.ok_or(StatusCode::BAD_REQUEST)
}
#[derive(Deserialize)]
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
pub struct DebugSeedEventsRequest {
pub partition: String,
pub count: u64,
}
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
pub async fn handle_debug_seed_events(
State(state): State<Arc<AppState>>,
Query(req): Query<DebugSeedEventsRequest>,
) -> Result<StatusCode, StatusCode> {
if req.partition != "events" && req.partition != "relay_events" {
return Err(StatusCode::BAD_REQUEST);
}
let count = seed_count(req.count)?;
state
.db
.run(move |db| {
let mut txn = crate::db::Txn::new(db);
if req.partition == "events" {
#[cfg(feature = "indexer_stream")]
for _ in 0..count {
txn.outbox.stream.push_record(b"dummy".to_vec(), None);
}
} else if req.partition == "relay_events" {
// real frames, because subscribeRepos clients get these rows
// live and on replay like any other
#[cfg(feature = "relay")]
{
use crate::ingest::stream::{Datetime, Identity};
let did = jacquard_common::types::string::Did::new_static(
"did:plc:aaaaaaaaaaaaaaaaaaaaaaaa",
)
.into_diagnostic()?;
let time =
Datetime::try_from("2026-08-10T12:34:56Z".to_owned()).into_diagnostic()?;
for _ in 0..count {
txn.outbox.relay.push_identity(Identity {
did: did.clone(),
handle: None,
seq: 0,
time: time.clone(),
});
}
}
}
txn.commit()
})
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(StatusCode::OK)
}
#[cfg(all(feature = "relay", feature = "jetstream"))]
#[derive(Deserialize)]
pub struct DebugSeedRelayJetstreamCommitsRequest {
pub count: u64,
}
#[cfg(all(feature = "relay", feature = "jetstream"))]
pub async fn handle_debug_seed_relay_jetstream_commits(
State(state): State<Arc<AppState>>,
Query(req): Query<DebugSeedRelayJetstreamCommitsRequest>,
) -> Result<StatusCode, StatusCode> {
let count = seed_count(req.count)?;
state
.db
.run(move |db| {
use crate::db::types::TrimmedDid;
use crate::ingest::stream::{
Account, AccountStatus, Commit, Datetime, Identity, RepoOp, RepoOpAction,
SubscribeReposMessage, decode_frame, encode_frame,
};
use crate::types::StoredJetstreamEvent;
use jacquard_common::CowStr;
use jacquard_common::types::cid::CidLink;
use jacquard_common::types::string::{Did, Handle, Tid};
// each frame is encoded at its seq when the outbox commits. the
// local encoding here only checks the fixture survives the codec.
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").into_diagnostic()?;
let commit_cid = CidLink::from(crate::car::compute_cid(b"relay-jetstream-fixture"));
let rev: Tid = "3kzbif5moe22m".parse().into_diagnostic()?;
let time = Datetime::try_from("2026-08-10T12:34:56Z".to_owned()).into_diagnostic()?;
let trimmed_did = TrimmedDid::from(&did).into_static();
let mut txn = crate::db::Txn::new(db);
for index in 0..count {
let rkey = format!("fixture{index}");
let commit = Commit {
blobs: Vec::new(),
blocks: bytes::Bytes::new(),
commit: commit_cid.clone(),
ops: vec![RepoOp {
action: RepoOpAction::Delete,
cid: None,
path: CowStr::Owned(format!("app.bsky.feed.post/{rkey}").into()),
prev: None,
}],
prev_data: None,
rebase: false,
repo: did.clone(),
rev: rev.clone(),
seq: 0,
since: None,
time: time.clone(),
too_big: false,
};
let frame = encode_frame("#commit", &commit)?;
let SubscribeReposMessage::Commit(decoded) =
decode_frame(frame.as_ref()).map_err(|err| miette::miette!("{err}"))?
else {
return Err(miette::miette!("relay fixture did not decode as a commit"));
};
if decoded.ops.len() != 1 {
return Err(miette::miette!("relay fixture lost its operation"));
}
let relay_seq = txn.outbox.relay.push_commit(commit);
txn.outbox.jetstream.push(
StoredJetstreamEvent::RelayCommit {
did: trimmed_did.clone(),
collection: CowStr::Borrowed("app.bsky.feed.post"),
relay_seq,
op_index: 0,
},
Some(crate::jetstream::JetstreamEphemeral {
did: did.as_str().to_owned(),
rev: rev.as_str().to_owned(),
operation: "delete".to_owned(),
collection: "app.bsky.feed.post".to_owned(),
rkey,
record: None,
cid: None,
live: true,
}),
);
}
let account = Account {
active: false,
did: did.clone(),
seq: 0,
status: Some(AccountStatus::Deactivated),
time: time.clone(),
};
let frame = encode_frame("#account", &account)?;
let SubscribeReposMessage::Account(decoded) =
decode_frame(frame.as_ref()).map_err(|err| miette::miette!("{err}"))?
else {
return Err(miette::miette!(
"relay fixture did not decode as an account"
));
};
if decoded.active || decoded.status != Some(AccountStatus::Deactivated) {
return Err(miette::miette!("relay account fixture lost its state"));
}
let relay_seq = txn.outbox.relay.push_account(account);
txn.outbox.jetstream.push(
StoredJetstreamEvent::RelayAccount {
did: trimmed_did.clone(),
relay_seq,
},
None,
);
let identity = Identity {
did: did.clone(),
handle: Some(Handle::new_static("alice.test").into_diagnostic()?),
seq: 0,
time: time.clone(),
};
let frame = encode_frame("#identity", &identity)?;
let SubscribeReposMessage::Identity(decoded) =
decode_frame(frame.as_ref()).map_err(|err| miette::miette!("{err}"))?
else {
return Err(miette::miette!(
"relay fixture did not decode as an identity"
));
};
if decoded.handle.as_ref().map(Handle::as_str) != Some("alice.test") {
return Err(miette::miette!("relay identity fixture lost its handle"));
}
let relay_seq = txn.outbox.relay.push_identity(identity);
txn.outbox.jetstream.push(
StoredJetstreamEvent::RelayIdentity {
did: trimmed_did,
relay_seq,
},
None,
);
txn.commit()
})
.await
.map_err(|err| {
tracing::error!(%err, "failed to seed relay Jetstream fixture");
StatusCode::INTERNAL_SERVER_ERROR
})?;
Ok(StatusCode::OK)
}
#[cfg(all(feature = "indexer_stream", feature = "backlinks"))]
#[derive(Deserialize)]
pub struct DebugSeedBacklinksRequest {
pub like_count: u64,
pub post_count: u64,
}
#[cfg(all(feature = "indexer_stream", feature = "backlinks"))]
#[derive(Serialize)]
pub struct DebugSeedBacklinksResponse {
pub did: String,
pub subject: String,
pub like_count: u64,
pub post_count: u64,
}
#[cfg(all(feature = "indexer_stream", feature = "backlinks"))]
pub async fn handle_debug_seed_backlinks(
State(state): State<Arc<AppState>>,
Query(req): Query<DebugSeedBacklinksRequest>,
) -> Result<Json<DebugSeedBacklinksResponse>, StatusCode> {
req.like_count
.checked_add(req.post_count)
.filter(|count| *count <= 10_000)
.ok_or(StatusCode::BAD_REQUEST)?;
state
.db
.run(move |db| {
use crate::db::types::DbRkey;
use jacquard_common::types::string::Did;
const DID: &str = "did:plc:wqstj3k5tslmm246baaf3tpa";
const SUBJECT: &str = "at://did:plc:ewvi7nxzyoun6zhxrhs64oiz/app.bsky.feed.post/root";
const SUBJECT_CID: &str = "bafyreicldnax5mfobeba2pnv7fgsdfz2563gc72gyi5cnhyuqfnk3jz7aa";
let did = Did::new_static(DID).into_diagnostic()?;
let mut batch = db.inner.batch();
for (collection, count) in [
("app.bsky.feed.like", req.like_count),
("app.bsky.feed.post", req.post_count),
] {
for index in 0..count {
let rkey = if collection == "app.bsky.feed.like" && index == 0 {
"stinkpot".to_owned()
} else {
let prefix = if collection == "app.bsky.feed.like" {
"like"
} else {
"post"
};
format!("{prefix}{index:04}")
};
let body = serde_ipld_dagcbor::to_vec(&serde_json::json!({
"$type": collection,
"subject": {
"uri": SUBJECT,
"cid": SUBJECT_CID,
},
"createdAt": "2026-08-10T12:34:56Z",
}))
.into_diagnostic()?;
let value = serde_ipld_dagcbor::from_slice::<jacquard_common::Data>(&body)
.into_diagnostic()?;
let record_key =
crate::db::keys::record_key(&did, collection, &DbRkey::new(&rkey));
db.indexer
.stage_head(&mut batch, record_key, body.as_slice());
crate::backlinks::store::index_record(
&mut batch,
&db.backlinks,
did.as_str(),
collection,
&rkey,
&value,
)?;
}
}
batch.commit().into_diagnostic()?;
Ok::<_, miette::Report>(DebugSeedBacklinksResponse {
did: DID.to_owned(),
subject: SUBJECT.to_owned(),
like_count: req.like_count,
post_count: req.post_count,
})
})
.await
.map(Json)
.map_err(|err| {
tracing::error!(%err, "failed to seed backlinks fixture");
StatusCode::INTERNAL_SERVER_ERROR
})
}
#[cfg(all(test, any(feature = "indexer_stream", feature = "relay")))]
mod tests {
use super::*;
/// every seeded relay row has to be a frame, since subscribeRepos sends
/// the rows to its clients as they are, live and on replay
#[cfg(feature = "relay")]
#[tokio::test]
async fn seeded_relay_rows_reach_subscribers_as_frames() -> miette::Result<()> {
use futures::StreamExt;
let tmp = tempfile::tempdir().into_diagnostic()?;
let hydrant = crate::control::Hydrant::new(crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})
.await?;
let live = hydrant.subscribe_repos(None);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.into_diagnostic()?;
let addr = listener.local_addr().into_diagnostic()?;
let app = router().with_state(hydrant.state.clone());
tokio::spawn(async move { axum::serve(listener, app).await });
let response = reqwest::Client::new()
.post(format!(
"http://{addr}/debug/seed_events?partition=relay_events&count=3"
))
.send()
.await
.into_diagnostic()?;
assert_eq!(response.status(), reqwest::StatusCode::OK);
for (name, stream) in [("live", live), ("replay", hydrant.subscribe_repos(Some(0)))] {
let frames = tokio::time::timeout(
std::time::Duration::from_secs(10),
stream.take(3).collect::<Vec<_>>(),
)
.await
.into_diagnostic()?;
for frame in frames {
let frame = frame.into_diagnostic()?;
crate::ingest::stream::decode_frame(&frame)
.map_err(|err| miette::miette!("{name} subscriber got a non-frame: {err}"))?;
}
}
Ok(())
}
#[test]
fn seeds_are_capped() {
assert_eq!(seed_count(0), Ok(0));
assert_eq!(seed_count(10_000), Ok(10_000));
assert_eq!(seed_count(10_001), Err(StatusCode::BAD_REQUEST));
assert_eq!(seed_count(u64::MAX), Err(StatusCode::BAD_REQUEST));
}
}
#[cfg(all(test, feature = "indexer"))]
mod key_round_trip_tests {
use super::*;
use crate::db::types::{DbRkey, DbTid};
use jacquard_common::types::string::Did;
#[tokio::test]
async fn history_keys_from_iter_work_in_get_and_bounds() -> miette::Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let hydrant = crate::control::Hydrant::new(crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})
.await?;
let did = Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").into_diagnostic()?;
let history = hydrant
.state
.db
.keyspace_by_name("history")
.expect("history is open");
let record = |rkey: &str| keys::record_key(&did, "app.bsky.feed.post", &DbRkey::new(rkey));
// the last rev has its top bit set, which no real tid has, but the key
// still has to come back byte for byte
let rows = [
(record("3kzbif5moe22m"), [0, 0, 0, 0, 0, 0, 4, 0]),
(record("3kzbif5moe22m"), [0, 0, 0, 0, 0, 0, 8, 0]),
(record("self"), [0xff; 8]),
];
for (index, (record, rev)) in rows.iter().enumerate() {
let body = serde_ipld_dagcbor::to_vec(&serde_json::json!({ "index": index }))
.into_diagnostic()?;
history
.insert(
keys::history_key(record, &DbTid::new_from_bytes(*rev)),
body,
)
.into_diagnostic()?;
}
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.into_diagnostic()?;
let addr = listener.local_addr().into_diagnostic()?;
let app = router().with_state(hydrant.state.clone());
tokio::spawn(async move { axum::serve(listener, app).await });
let client = reqwest::Client::new();
let get = |path: &'static str, query: Vec<(&'static str, String)>| {
let request = client.get(format!("http://{addr}{path}")).query(&query);
async move {
let response = request.send().await.into_diagnostic()?;
assert_eq!(
response.status(),
reqwest::StatusCode::OK,
"{path} {query:?}"
);
response.json::<Value>().await.into_diagnostic()
}
};
let iter = |bounds: &[(&'static str, &str)]| {
let mut query = vec![("partition", "history".to_owned())];
query.extend(bounds.iter().map(|(name, key)| (*name, key.to_string())));
get("/debug/iter", query)
};
let all = iter(&[]).await?["items"].clone();
let items = all.as_array().expect("items is a list");
assert_eq!(items.len(), rows.len());
for item in items {
let key = item[0].as_str().expect("keys render as strings");
let got = get(
"/debug/get",
vec![("partition", "history".into()), ("key", key.into())],
)
.await?;
assert_eq!(got["value"], item[1], "/debug/get {key}");
}
let middle = items[1][0].as_str().expect("keys render as strings");
let one = iter(&[("start", middle), ("end", middle)]).await?;
assert_eq!(one["items"], Value::Array(vec![items[1].clone()]));
let rest = iter(&[("start", middle)]).await?;
assert_eq!(rest["items"], Value::Array(items[1..].to_vec()));
Ok(())
}
}