use std::collections::HashMap;
use axum::{Json, Router, extract::State, http::StatusCode, routing::get};
use jacquard_common::types::string::Did;
use serde::{Deserialize, Serialize};
use smol_str::SmolStr;
use crate::{api::MultiQuery, control::Hydrant};
pub fn router() -> Router<Hydrant> {
Router::new()
.route(
"/xrpc/blue.microcosm.links.getBacklinks",
get(handle_get_backlinks),
)
.route(
"/xrpc/blue.microcosm.links.getBacklinksCount",
get(handle_get_backlinks_count),
)
}
#[derive(Deserialize)]
pub struct GetBacklinksParams {
pub subject: String,
/// filter by source collection, optionally with a path suffix `collection:path`
pub source: Option<String>,
#[serde(default)]
pub did: Vec<String>,
pub limit: Option<u64>,
pub cursor: Option<String>,
pub reverse: Option<bool>,
}
#[derive(Serialize)]
pub struct Backlink {
pub uri: SmolStr,
pub cid: SmolStr,
}
#[derive(Serialize)]
pub struct GetBacklinksOutput {
pub backlinks: Vec<Backlink>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cursor: Option<String>,
}
#[derive(Deserialize)]
pub struct GetBacklinksCountParams {
pub subject: String,
pub source: Option<String>,
#[serde(default)]
pub did: Vec<String>,
}
#[derive(Serialize)]
pub struct GetBacklinksCountOutput {
pub count: u64,
}
pub async fn handle_get_backlinks(
State(hydrant): State<Hydrant>,
MultiQuery(params): MultiQuery<GetBacklinksParams>,
) -> Result<Json<GetBacklinksOutput>, StatusCode> {
let limit = params.limit.unwrap_or(50).min(100) as usize;
let reverse = params.reverse.unwrap_or(false);
let cursor_bytes = params
.cursor
.as_deref()
.and_then(|c| data_encoding::BASE64URL_NOPAD.decode(c.as_bytes()).ok());
let mut fetch = hydrant
.backlinks
.fetch(params.subject)
.limit(limit)
.reverse(reverse);
if !params.did.is_empty() {
fetch = fetch.dids(params.did);
}
if let Some(ref src) = params.source {
fetch = fetch.source(src);
}
if let Some(c) = cursor_bytes {
fetch = fetch.cursor(c);
}
let page = fetch
.run()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let next_cursor = page
.next_cursor
.map(|b| data_encoding::BASE64URL_NOPAD.encode(&b));
// a link is a record of the linking account, which isn't served while that account is
// inactive, so a page can come back shorter than the limit
let mut served = HashMap::<SmolStr, bool>::new();
let mut backlinks = Vec::with_capacity(page.backlinks.len());
for e in page.backlinks {
let Some(did) = e
.uri
.strip_prefix("at://")
.and_then(|rest| rest.split('/').next())
else {
continue;
};
let serve = match served.get(did) {
Some(serve) => *serve,
None => {
let serve = match Did::new_owned(did) {
Ok(did) => crate::api::unavailable(&hydrant.repos.get(&did))
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.is_none(),
Err(_) => false,
};
served.insert(did.into(), serve);
serve
}
};
if serve {
backlinks.push(Backlink {
uri: e.uri,
cid: e.cid,
});
}
}
Ok(Json(GetBacklinksOutput {
backlinks,
cursor: next_cursor,
}))
}
pub async fn handle_get_backlinks_count(
State(hydrant): State<Hydrant>,
MultiQuery(params): MultiQuery<GetBacklinksCountParams>,
) -> Result<Json<GetBacklinksCountOutput>, StatusCode> {
let mut count_q = hydrant.backlinks.count(params.subject);
if !params.did.is_empty() {
count_q = count_q.dids(params.did);
}
if let Some(ref src) = params.source {
count_q = count_q.source(src);
}
let count = count_q
.run()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(Json(GetBacklinksCountOutput { count }))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::Config;
use crate::db::{self, keys, types::DbRkey};
use crate::types::{RepoState, RepoStatus};
use miette::IntoDiagnostic;
const SUBJECT: &str = "at://did:plc:target/app.bsky.feed.post/3jz";
/// a hydrant where a like by a repo whose row says `status` links to `SUBJECT`, the like
/// still in place, like a deleted account whose wipe hasn't run yet
async fn liked_by(status: RepoStatus) -> miette::Result<(tempfile::TempDir, Hydrant)> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let hydrant = Hydrant::new(Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})
.await?;
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
let value = serde_json::json!({
"$type": "app.bsky.feed.like",
"subject": {
"uri": SUBJECT,
"cid": "bafyreig2toyab2fv7kqv3wcldlgl63p2oqxv54htv23s3kbsnshs7j2yli"
},
"createdAt": "2026-08-04T00:00:00Z"
});
let body = serde_ipld_dagcbor::to_vec(&value).into_diagnostic()?;
let data = serde_ipld_dagcbor::from_slice(&body).into_diagnostic()?;
let mut repo = RepoState::backfilling();
repo.active = status == RepoStatus::Synced;
repo.status = status;
let db = &hydrant.state.db;
let mut batch = db.inner.batch();
batch.insert(&db.repos, keys::repo_key(&did), db::ser_repo_state(&repo)?);
db.indexer.stage_head(
&mut batch,
keys::record_key(&did, "app.bsky.feed.like", &DbRkey::new("3kz")),
body,
);
crate::backlinks::store::index_record(
&mut batch,
&db.backlinks,
did.as_str(),
"app.bsky.feed.like",
"3kz",
&data,
)?;
batch.commit().into_diagnostic()?;
Ok((tmp, hydrant))
}
async fn backlinks(hydrant: &Hydrant) -> usize {
let params = GetBacklinksParams {
subject: SUBJECT.into(),
source: None,
did: Vec::new(),
limit: None,
cursor: None,
reverse: None,
};
let Json(out) = handle_get_backlinks(State(hydrant.clone()), MultiQuery(params))
.await
.unwrap();
out.backlinks.len()
}
#[tokio::test]
async fn an_inactive_accounts_links_are_not_served() -> miette::Result<()> {
let (_tmp, hydrant) = liked_by(RepoStatus::Synced).await?;
assert_eq!(backlinks(&hydrant).await, 1);
for status in [
RepoStatus::Deleted,
RepoStatus::Takendown,
RepoStatus::Suspended,
RepoStatus::Deactivated,
] {
let (_tmp, hydrant) = liked_by(status.clone()).await?;
assert_eq!(backlinks(&hydrant).await, 0, "{status:?}");
}
Ok(())
}
#[test]
fn backlinks_query_accepts_single_and_repeated_dids() {
let fetch: GetBacklinksParams = serde_html_form::from_str(
"subject=at%3A%2F%2Ftarget&did=did%3Aplc%3Aone&did=did%3Aweb%3Atwo",
)
.unwrap();
assert_eq!(fetch.did, ["did:plc:one", "did:web:two"]);
let count: GetBacklinksCountParams =
serde_html_form::from_str("subject=at%3A%2F%2Ftarget&did=did%3Aplc%3Aone").unwrap();
assert_eq!(count.did, ["did:plc:one"]);
}
}