use async_trait::async_trait;
use url::Url;
use crate::provenance::{Capability, LogEvent, LogResult, RowInput};
use crate::source::{FetchContext, FetchError, FetchResult, Source};
use crate::{CapabilityProfile, Ref};
const DEFAULT_BASE: &str = "https://api.openalex.org";
#[derive(Clone, Debug)]
pub struct OpenalexSource {
base: Url,
contact_email: String,
}
impl OpenalexSource {
#[must_use]
pub fn new(contact_email: String) -> Self {
Self {
#[allow(clippy::expect_used)]
base: Url::parse(DEFAULT_BASE).expect("hard-coded base URL is valid"),
contact_email,
}
}
pub fn with_base(base: Url, contact_email: String) -> Self {
Self {
base,
contact_email,
}
}
fn request_url(&self, doi: &crate::Doi) -> Result<Url, FetchError> {
let path = format!("/works/doi:{}", doi.as_str());
let mut url = self
.base
.join(&path)
.map_err(|e| FetchError::SourceSchema {
hint: format!("openalex URL construction failed: {e}"),
})?;
if !self.contact_email.is_empty() {
url.query_pairs_mut()
.append_pair("mailto", &self.contact_email);
}
Ok(url)
}
}
#[async_trait]
impl Source for OpenalexSource {
fn name(&self) -> &str {
"openalex"
}
fn can_serve(&self, profile: &CapabilityProfile, ref_: &Ref) -> bool {
profile.metadata.openalex && matches!(ref_, Ref::Doi(_))
}
async fn fetch(
&self,
ref_: &Ref,
profile: &CapabilityProfile,
ctx: &FetchContext,
) -> Result<FetchResult, FetchError> {
let doi = match ref_ {
Ref::Doi(d) => d,
Ref::Arxiv(_) => {
return Err(FetchError::NotEligible {
source_key: "openalex".into(),
});
}
};
if !profile.metadata.openalex {
return Err(FetchError::NotEligible {
source_key: "openalex".into(),
});
}
let _permit = ctx.rate_limiter.acquire(self.name()).await;
let url = self.request_url(doi)?;
let (body, final_url) = ctx.http.fetch_bytes(self.name(), url).await?;
let work: serde_json::Value =
serde_json::from_slice(&body).map_err(|e| FetchError::SourceSchema {
hint: format!("openalex returned non-JSON: {e}"),
})?;
if work.get("id").is_none() {
return Err(FetchError::SourceSchema {
hint: format!(
"openalex response missing `id` field — likely an error \
payload (got: {})",
truncate_for_hint(&body)
),
});
}
let canonical = ref_.promote(self.name(), None).digest_hex();
ctx.log.append(RowInput {
event: LogEvent::Fetch,
result: LogResult::Ok,
capability: Capability::Metadata,
ref_: Some(doi.as_str()),
source: Some(self.name()),
error_code: None,
size_bytes: Some(body.len() as u64),
license: None,
store_path: None,
canonical_digest: Some(&canonical),
})?;
Ok(FetchResult {
source: self.name().to_string(),
license: "unknown".into(),
pdf_bytes: None,
final_url: Some(final_url),
metadata_json: Some(work),
})
}
}
fn truncate_for_hint(body: &[u8]) -> String {
const MAX: usize = 200;
let s = String::from_utf8_lossy(body);
if s.len() <= MAX {
s.into_owned()
} else {
format!("{}…", &s[..MAX])
}
}
#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod tests {
use super::*;
use std::sync::Arc;
use camino::Utf8PathBuf;
use tempfile::TempDir;
use wiremock::matchers::{method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate};
use crate::http::HttpClient;
use crate::provenance::ProvenanceLog;
use crate::rate_limiter::RateLimiter;
use crate::{ArxivId, CapabilityProfile, Doi, MetadataAccess, RateLimits, Ref};
const SAMPLE_WORK: &str = r#"{
"id": "https://openalex.org/W2741809807",
"doi": "https://doi.org/10.1234/example",
"display_name": "Example Work Title",
"publication_year": 2024,
"referenced_works": [
"https://openalex.org/W2000000001",
"https://openalex.org/W2000000002"
]
}"#;
fn build_test_context(wiremock_host: &str) -> (TempDir, FetchContext) {
let td = TempDir::new().expect("tempdir");
let log_dir =
Utf8PathBuf::try_from(td.path().to_path_buf()).expect("temp dir path must be UTF-8");
let log_path = log_dir.join("test.jsonl");
let http = Arc::new(HttpClient::new_for_tests_allow_http(
"openalex",
wiremock_host,
));
let rate_limiter = Arc::new(RateLimiter::new(RateLimits::HARD_CODED));
let session_id = "01J0000000000000000000TEST".to_string();
let log = Arc::new(
ProvenanceLog::open(log_path, session_id.clone()).expect("provenance log opens"),
);
let ctx = FetchContext {
http,
rate_limiter,
log,
session_id,
};
(td, ctx)
}
fn profile_with_openalex_enabled() -> CapabilityProfile {
let mut p = CapabilityProfile::from_env().expect("clean env never errors");
p.metadata = MetadataAccess {
openalex: true,
semantic_scholar: false,
doaj: false,
};
p
}
#[tokio::test]
async fn fetch_doi_returns_work_metadata() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/works/doi:10.1234/example"))
.and(query_param("mailto", "doiget@localhost"))
.respond_with(ResponseTemplate::new(200).set_body_string(SAMPLE_WORK))
.mount(&server)
.await;
let (_td, ctx) = build_test_context(&server.uri());
let src = OpenalexSource::with_base(
Url::parse(&server.uri()).expect("wiremock URI parses"),
"doiget@localhost".to_string(),
);
let profile = profile_with_openalex_enabled();
let ref_ = Ref::Doi(Doi::parse("10.1234/example").expect("DOI parses"));
let result = src.fetch(&ref_, &profile, &ctx).await.expect("fetch ok");
assert_eq!(result.source, "openalex");
assert!(result.pdf_bytes.is_none(), "metadata-only contract");
let meta = result.metadata_json.expect("metadata_json present");
assert_eq!(meta["display_name"], "Example Work Title");
assert_eq!(
meta["referenced_works"][0],
"https://openalex.org/W2000000001"
);
}
#[tokio::test]
async fn fetch_arxiv_id_is_not_eligible() {
let (_td, ctx) = build_test_context("http://127.0.0.1:1");
let src = OpenalexSource::with_base(
Url::parse("http://127.0.0.1:1").expect("URI parses"),
"doiget@localhost".to_string(),
);
let profile = profile_with_openalex_enabled();
let ref_ = Ref::Arxiv(ArxivId::parse("2401.12345").expect("arXiv id parses"));
let err = src
.fetch(&ref_, &profile, &ctx)
.await
.expect_err("arXiv ref must be rejected");
assert!(matches!(err, FetchError::NotEligible { .. }));
}
#[tokio::test]
async fn fetch_without_capability_flag_is_not_eligible() {
let (_td, ctx) = build_test_context("http://127.0.0.1:1");
let src = OpenalexSource::with_base(
Url::parse("http://127.0.0.1:1").expect("URI parses"),
"doiget@localhost".to_string(),
);
let profile = CapabilityProfile::from_env().expect("clean env never errors");
let ref_ = Ref::Doi(Doi::parse("10.1234/example").expect("DOI parses"));
assert!(
!src.can_serve(&profile, &ref_),
"can_serve must be false without DOIGET_ENABLE_OPENALEX"
);
let err = src
.fetch(&ref_, &profile, &ctx)
.await
.expect_err("fetch must reject when capability is denied");
assert!(matches!(err, FetchError::NotEligible { .. }));
}
#[tokio::test]
async fn fetch_malformed_response_returns_source_schema_error() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/works/doi:10.1234/example"))
.respond_with(ResponseTemplate::new(200).set_body_string(r#"{"error":"not found"}"#))
.mount(&server)
.await;
let (_td, ctx) = build_test_context(&server.uri());
let src = OpenalexSource::with_base(
Url::parse(&server.uri()).expect("wiremock URI parses"),
"doiget@localhost".to_string(),
);
let profile = profile_with_openalex_enabled();
let ref_ = Ref::Doi(Doi::parse("10.1234/example").expect("DOI parses"));
let err = src
.fetch(&ref_, &profile, &ctx)
.await
.expect_err("missing `id` must surface as SourceSchema");
assert!(matches!(err, FetchError::SourceSchema { .. }));
}
}