From a17be94ffc51261cd03d99c3bf1a5a1e50f793b3 Mon Sep 17 00:00:00 2001 From: James Kane Date: Sat, 1 Aug 2026 06:06:43 -0500 Subject: [PATCH] fix(pubs): carry open-access metadata through the discovery review queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A paper promoted from the OpenAlex discovery queue landed on /references with no "Open access" badge and no citation count, until the nightly by-DOI enrichment job happened to re-fetch it. The search response already carries `open_access.oa_status` and `cited_by_count` — the wire type even parsed them — but `Candidate` dropped both, `pubs.publication_candidate` had nowhere to put them, and `promote_candidate`'s INSERT omitted them. Carry them end to end (discovery job and the public "suggest a paper" form), and gap-fill rather than overwrite when promotion reuses an existing publication. Two related defects on the same path: - Publications with an OpenAlex id but no DOI were unreachable by enrichment — the work-list is `WHERE doi IS NOT NULL`, and a candidate is promoted on its OpenAlex id alone. Adds `work_by_id` and a second pass over those rows. - OpenAlex returns `doi` as a resolver URL, stored verbatim. That rendered the reference list's link as https://doi.org/https://doi.org/... and made `exists_by_doi` miss, so a paper already in the catalog could be queued a second time from the public form. Ingest now normalizes to bare form and mig 0073 backfills what is stored (30 publications, 5,934 candidates in dev). `upsert_candidate` takes a `NewCandidate` struct — the positional list was at seven arguments and this would have made nine. Co-Authored-By: Claude Opus 5 (1M context) --- rust/crates/du-db/src/publication.rs | 95 ++++++++++++++----- .../du-db/tests/publication_candidate.rs | 30 +++++- rust/crates/du-external/src/openalex.rs | 72 +++++++++++++- rust/crates/du-jobs/src/publications.rs | 37 ++++++-- rust/crates/du-web/src/routes/references.rs | 27 +++--- .../0073_pubs_candidate_openaccess.sql | 34 +++++++ 6 files changed, 244 insertions(+), 51 deletions(-) create mode 100644 rust/migrations/0073_pubs_candidate_openaccess.sql diff --git a/rust/crates/du-db/src/publication.rs b/rust/crates/du-db/src/publication.rs index ff33c3c..1df245f 100644 --- a/rust/crates/du-db/src/publication.rs +++ b/rust/crates/du-db/src/publication.rs @@ -29,6 +29,19 @@ pub async fn dois(pool: &PgPool) -> Result, DbError Ok(rows.into_iter().map(|(id, doi)| (PublicationId(id), doi)).collect()) } +/// Publications the by-DOI work-list can never reach: an OpenAlex id but no DOI. +/// A discovery candidate is promoted on its OpenAlex id alone, so without this +/// pass those rows keep NULL citations / open-access status forever. +pub async fn openalex_ids_without_doi(pool: &PgPool) -> Result, DbError> { + let rows: Vec<(i64, String)> = sqlx::query_as( + "SELECT id, open_alex_id FROM pubs.publication \ + WHERE open_alex_id IS NOT NULL AND doi IS NULL ORDER BY id", + ) + .fetch_all(pool) + .await?; + Ok(rows.into_iter().map(|(id, oa)| (PublicationId(id), oa)).collect()) +} + /// Apply OpenAlex enrichment (only overwrites a column when the new value is set). pub async fn update_openalex(pool: &PgPool, id: PublicationId, u: &OpenAlexUpdate) -> Result { let affected = sqlx::query( @@ -140,34 +153,44 @@ pub async fn enabled_search_configs(pool: &PgPool) -> Result, .collect()) } +/// The metadata a discovery candidate is written with. `doi` is expected in bare +/// form (`du_external::openalex::normalize_doi`) — the catalog stores DOIs bare. +#[derive(Debug, Default, Clone)] +pub struct NewCandidate<'a> { + pub openalex_id: &'a str, + pub doi: Option<&'a str>, + pub title: Option<&'a str>, + pub abstract_summary: Option<&'a str>, + pub publication_date: Option, + pub journal_name: Option<&'a str>, + pub cited_by_count: Option, + pub open_access_status: Option<&'a str>, +} + /// Upsert a discovery candidate by OpenAlex id (preserves curator status/review). /// Returns `true` when the row was newly inserted (vs an update of one already seen). -#[allow(clippy::too_many_arguments)] -pub async fn upsert_candidate( - pool: &PgPool, - openalex_id: &str, - doi: Option<&str>, - title: Option<&str>, - abstract_summary: Option<&str>, - publication_date: Option, - journal_name: Option<&str>, -) -> Result { +pub async fn upsert_candidate(pool: &PgPool, c: &NewCandidate<'_>) -> Result { // `xmax = 0` distinguishes a fresh INSERT from a DO UPDATE on conflict. let inserted: bool = sqlx::query_scalar( "INSERT INTO pubs.publication_candidate \ - (openalex_id, doi, title, abstract, publication_date, journal_name, status) \ - VALUES ($1, $2, $3, $4, $5, $6, 'pending') \ + (openalex_id, doi, title, abstract, publication_date, journal_name, \ + cited_by_count, open_access_status, status) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'pending') \ ON CONFLICT (openalex_id) DO UPDATE SET doi = EXCLUDED.doi, title = EXCLUDED.title, \ abstract = EXCLUDED.abstract, publication_date = EXCLUDED.publication_date, \ - journal_name = EXCLUDED.journal_name \ + journal_name = EXCLUDED.journal_name, \ + cited_by_count = COALESCE(EXCLUDED.cited_by_count, pubs.publication_candidate.cited_by_count), \ + open_access_status = COALESCE(EXCLUDED.open_access_status, pubs.publication_candidate.open_access_status) \ RETURNING (xmax = 0)", ) - .bind(openalex_id) - .bind(doi) - .bind(title) - .bind(abstract_summary) - .bind(publication_date) - .bind(journal_name) + .bind(c.openalex_id) + .bind(c.doi) + .bind(c.title) + .bind(c.abstract_summary) + .bind(c.publication_date) + .bind(c.journal_name) + .bind(c.cited_by_count) + .bind(c.open_access_status) .fetch_one(pool) .await?; Ok(inserted) @@ -321,12 +344,15 @@ pub struct Candidate { pub publication_date: Option, pub journal_name: Option, pub relevance_score: Option, + pub cited_by_count: Option, + pub open_access_status: Option, pub status: String, pub created_at: DateTime, } const CAND_COLS: &str = "id, openalex_id, doi, title, abstract AS abstract_text, publication_date, \ - journal_name, relevance_score::float8 AS relevance_score, status, created_at"; + journal_name, relevance_score::float8 AS relevance_score, cited_by_count, open_access_status, \ + status, created_at"; /// Paginated candidate queue, optionally filtered by status, newest first. pub async fn list_candidates( @@ -385,6 +411,10 @@ pub async fn review_candidate( /// publication matching the candidate's OpenAlex id or DOI, else create one from /// the candidate's metadata; then mark the candidate `accepted`. Returns the /// publication id. Errors if the candidate has no title (publications require one). +/// +/// Citation count and open-access status come across too, so the promoted paper +/// carries its "Open access" badge immediately rather than from the next nightly +/// enrichment run. On reuse they gap-fill an existing row without overwriting it. pub async fn promote_candidate(pool: &PgPool, id: i64, by: Uuid) -> Result { let mut tx = pool.begin().await?; let c: Candidate = sqlx::query_as(&format!( @@ -410,11 +440,28 @@ pub async fn promote_candidate(pool: &PgPool, id: i64, by: Uuid) -> Result pid, + Some(pid) => { + // Gap-fill only: a publication the curators already own keeps its own + // values; this just supplies what the candidate knows and it doesn't. + sqlx::query( + "UPDATE pubs.publication SET \ + cited_by_count = COALESCE(cited_by_count, $2), \ + open_access_status = COALESCE(open_access_status, $3), \ + updated_at = now() \ + WHERE id = $1 AND (cited_by_count IS NULL OR open_access_status IS NULL)", + ) + .bind(pid) + .bind(c.cited_by_count) + .bind(c.open_access_status.as_deref()) + .execute(&mut *tx) + .await?; + pid + } None => { sqlx::query_scalar( - "INSERT INTO pubs.publication (open_alex_id, doi, title, journal, publication_date, abstract_summary) \ - VALUES ($1, $2, $3, $4, $5, $6) RETURNING id", + "INSERT INTO pubs.publication (open_alex_id, doi, title, journal, publication_date, \ + abstract_summary, cited_by_count, open_access_status) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id", ) .bind(&c.openalex_id) .bind(c.doi.as_deref()) @@ -422,6 +469,8 @@ pub async fn promote_candidate(pool: &PgPool, id: i64, by: Uuid) -> Result, } -/// A discovered candidate publication (from a search). +/// A discovered candidate publication (from a search). Carries the same +/// open-access / citation metadata as [`WorkMeta`] so a promoted candidate is a +/// fully-formed publication without waiting for the nightly by-DOI enrichment. #[derive(Debug, Clone, PartialEq)] pub struct Candidate { pub openalex_id: String, @@ -32,6 +34,8 @@ pub struct Candidate { pub abstract_summary: Option, pub publication_date: Option, pub journal: Option, + pub cited_by_count: Option, + pub open_access_status: Option, } // ── wire types ──────────────────────────────────────────────────────────────── @@ -102,6 +106,21 @@ fn short_id(url: &str) -> &str { url.rsplit('/').next().unwrap_or(url) } +/// Reduce a DOI to bare form (`10.1234/x`), stripping the `doi.org` resolver +/// prefix OpenAlex wraps around it and the `doi:` scheme people paste. Bare form +/// is what the catalog stores: the reference list builds its link as +/// `https://doi.org/{doi}`, and duplicate detection compares DOIs literally. +pub fn normalize_doi(raw: &str) -> String { + let d = raw.trim(); + let lower = d.to_ascii_lowercase(); + for prefix in ["https://doi.org/", "http://doi.org/", "https://dx.doi.org/", "http://dx.doi.org/", "doi:"] { + if lower.starts_with(prefix) { + return d[prefix.len()..].trim().to_string(); + } + } + d.to_string() +} + fn parse_date(s: &Option) -> Option { s.as_deref().and_then(|d| NaiveDate::parse_from_str(d, "%Y-%m-%d").ok()) } @@ -138,11 +157,13 @@ impl Work { let openalex_id = self.id.clone()?; Some(Candidate { openalex_id, - doi: self.doi.clone(), + doi: self.doi.as_deref().map(normalize_doi).filter(|d| !d.is_empty()), title: self.title.clone(), abstract_summary: self.abstract_inverted_index.as_ref().and_then(reconstruct_abstract), publication_date: parse_date(&self.publication_date), journal: self.primary_location.and_then(|l| l.source).and_then(|s| s.display_name), + cited_by_count: self.cited_by_count.map(|c| c as i32), + open_access_status: self.open_access.and_then(|o| o.oa_status), }) } } @@ -173,6 +194,25 @@ impl OpenAlexClient { Ok(Some(work.into_meta())) } + /// Fetch enrichment metadata by OpenAlex work id (bare `W…` or the full + /// `https://openalex.org/W…` URL). The by-id counterpart of + /// [`work_by_doi`](Self::work_by_doi), for catalog rows that have an OpenAlex + /// id but no DOI — a promoted discovery candidate need not have one, and the + /// by-DOI job can never reach those. Returns None on 404. + pub async fn work_by_id(&self, openalex_id: &str) -> Result, ExternalError> { + let url = format!("{}/works/{}", self.base, short_id(openalex_id.trim())); + let mut req = self.http.get(url); + if let Some(m) = &self.mailto { + req = req.query(&[("mailto", m.as_str())]); + } + let resp = req.send().await?; + if resp.status() == reqwest::StatusCode::NOT_FOUND { + return Ok(None); + } + let work: Work = resp.error_for_status()?.json().await?; + Ok(Some(work.into_meta())) + } + /// Fetch a single work's primary-topic id (short form, e.g. `T10751`). Used by /// the topic-prune backfill to retroactively test the pending candidate queue /// against a config's `primary_topic.id` whitelist. `openalex_id` may be a bare @@ -296,5 +336,33 @@ mod tests { assert_eq!(cands.len(), 1); assert_eq!(cands[0].openalex_id, "https://openalex.org/W1"); assert_eq!(cands[0].title.as_deref(), Some("A")); + // The resolver prefix is stripped at ingest — the catalog stores bare DOIs. + assert_eq!(cands[0].doi.as_deref(), Some("10.1/a")); + } + + /// The search payload carries open-access + citation metadata; a candidate must + /// keep it, or a promoted paper shows no OA badge until the nightly job re-fetches. + #[test] + fn candidate_keeps_open_access_and_citations() { + let json = r#"{ + "id": "https://openalex.org/W9", + "doi": "https://doi.org/10.1/b", + "title": "B", + "cited_by_count": 7, + "open_access": { "oa_status": "gold" } + }"#; + let c = serde_json::from_str::(json).unwrap().into_candidate().unwrap(); + assert_eq!(c.cited_by_count, Some(7)); + assert_eq!(c.open_access_status.as_deref(), Some("gold")); + } + + #[test] + fn normalize_doi_strips_resolver_prefixes() { + assert_eq!(normalize_doi("https://doi.org/10.1002/advs.76320"), "10.1002/advs.76320"); + assert_eq!(normalize_doi("http://dx.doi.org/10.1/x"), "10.1/x"); + assert_eq!(normalize_doi("DOI:10.1/x"), "10.1/x"); + assert_eq!(normalize_doi(" 10.1/x "), "10.1/x"); + // A DOI is not a URL — anything else is left alone. + assert_eq!(normalize_doi("https://example.org/paper"), "https://example.org/paper"); } } diff --git a/rust/crates/du-jobs/src/publications.rs b/rust/crates/du-jobs/src/publications.rs index 169ad52..1b35e69 100644 --- a/rust/crates/du-jobs/src/publications.rs +++ b/rust/crates/du-jobs/src/publications.rs @@ -44,10 +44,13 @@ fn to_update(m: WorkMeta) -> OpenAlexUpdate { } } -/// Refresh OpenAlex metadata for every publication with a DOI. +/// Refresh OpenAlex metadata for every publication with a DOI, then for those with +/// only an OpenAlex id (promoted discovery candidates need not have a DOI — the +/// by-DOI pass can never reach them). pub async fn update_all(pool: &PgPool, client: &OpenAlexClient) -> anyhow::Result<()> { let dois = du_db::publication::dois(pool).await?; - let total = dois.len(); + let by_id = du_db::publication::openalex_ids_without_doi(pool).await?; + let total = dois.len() + by_id.len(); let (mut updated, mut missing, mut failed) = (0usize, 0usize, 0usize); for (id, doi) in dois { match client.work_by_doi(&doi).await { @@ -63,6 +66,20 @@ pub async fn update_all(pool: &PgPool, client: &OpenAlexClient) -> anyhow::Resul } tokio::time::sleep(REQUEST_GAP).await; } + for (id, openalex_id) in by_id { + match client.work_by_id(&openalex_id).await { + Ok(Some(meta)) => { + du_db::publication::update_openalex(pool, id, &to_update(meta)).await?; + updated += 1; + } + Ok(None) => missing += 1, + Err(e) => { + tracing::warn!(%openalex_id, error = %e, "openalex fetch by id failed"); + failed += 1; + } + } + tokio::time::sleep(REQUEST_GAP).await; + } tracing::info!(total, updated, missing, failed, "publication-update done"); Ok(()) } @@ -127,12 +144,16 @@ async fn run_search( for c in &page.candidates { let inserted = du_db::publication::upsert_candidate( pool, - &c.openalex_id, - c.doi.as_deref(), - c.title.as_deref(), - c.abstract_summary.as_deref(), - c.publication_date, - c.journal.as_deref(), + &du_db::publication::NewCandidate { + openalex_id: &c.openalex_id, + doi: c.doi.as_deref(), + title: c.title.as_deref(), + abstract_summary: c.abstract_summary.as_deref(), + publication_date: c.publication_date, + journal_name: c.journal.as_deref(), + cited_by_count: c.cited_by_count, + open_access_status: c.open_access_status.as_deref(), + }, ) .await?; seen += 1; diff --git a/rust/crates/du-web/src/routes/references.rs b/rust/crates/du-web/src/routes/references.rs index 83d5247..0a1883a 100644 --- a/rust/crates/du-web/src/routes/references.rs +++ b/rust/crates/du-web/src/routes/references.rs @@ -13,6 +13,7 @@ use axum::response::Response; use axum::routing::get; use axum::{Form, Router}; use du_domain::ids::PublicationId; +use du_external::openalex::normalize_doi; use serde::Deserialize; pub fn router() -> Router { @@ -251,13 +252,9 @@ struct SubmitForm { recaptcha: Option, } -/// Strip common DOI prefixes so OpenAlex's `/works/doi:` lookup gets a bare DOI. -fn normalize_doi(raw: &str) -> String { - let d = raw.trim(); - let d = d.strip_prefix("https://doi.org/").or_else(|| d.strip_prefix("http://doi.org/")).unwrap_or(d); - let d = d.strip_prefix("doi:").unwrap_or(d); - d.trim().to_string() -} +// DOIs are normalized to bare form (shared with the discovery ingest) so the +// OpenAlex `/works/doi:` lookup resolves and so `exists_by_doi` can match what +// the catalog stores. async fn submit_form(locale: Locale, user: MaybeUser) -> Response { html(&SubmitTemplate { @@ -321,12 +318,16 @@ async fn submit( du_db::publication::upsert_candidate( &st.pool, - openalex_id, - Some(&doi), - meta.title.as_deref(), - meta.abstract_summary.as_deref(), - meta.publication_date, - meta.journal.as_deref(), + &du_db::publication::NewCandidate { + openalex_id, + doi: Some(&doi), + title: meta.title.as_deref(), + abstract_summary: meta.abstract_summary.as_deref(), + publication_date: meta.publication_date, + journal_name: meta.journal.as_deref(), + cited_by_count: meta.cited_by_count, + open_access_status: meta.open_access_status.as_deref(), + }, ) .await?; diff --git a/rust/migrations/0073_pubs_candidate_openaccess.sql b/rust/migrations/0073_pubs_candidate_openaccess.sql new file mode 100644 index 0000000..738c1ba --- /dev/null +++ b/rust/migrations/0073_pubs_candidate_openaccess.sql @@ -0,0 +1,34 @@ +-- Carry OpenAlex open-access / citation metadata through the discovery review +-- queue, and store DOIs in bare form. +-- +-- The discovery search response already contains `open_access.oa_status` and +-- `cited_by_count`, but the candidate queue had nowhere to put them, so a +-- promoted candidate landed as a publication with NULL open_access_status — +-- no "Open access" badge and no citation count on /references until the nightly +-- by-DOI enrichment job happened to re-fetch the work. +-- +-- OpenAlex also returns `doi` as a resolver URL (https://doi.org/10.x). That was +-- stored verbatim, which (a) rendered the reference list's DOI link as +-- https://doi.org/https://doi.org/… and (b) made exists_by_doi() miss, so a +-- paper already in the catalog could be queued a second time from the public +-- "suggest a paper" form. Ingest now normalizes; this backfills what is stored. + +ALTER TABLE pubs.publication_candidate + ADD COLUMN cited_by_count INTEGER, + ADD COLUMN open_access_status TEXT; + +UPDATE pubs.publication_candidate + SET doi = regexp_replace(doi, '^(https?://(dx\.)?doi\.org/|doi:)', '', 'i') + WHERE doi ~* '^(https?://(dx\.)?doi\.org/|doi:)'; + +-- publication.doi is UNIQUE: skip any row whose bare form is already taken by +-- another publication (a pre-existing duplicate) rather than aborting the +-- migration. None exist in dev/cutover; this is a guard for prod. +UPDATE pubs.publication p + SET doi = regexp_replace(p.doi, '^(https?://(dx\.)?doi\.org/|doi:)', '', 'i') + WHERE p.doi ~* '^(https?://(dx\.)?doi\.org/|doi:)' + AND NOT EXISTS ( + SELECT 1 FROM pubs.publication o + WHERE o.id <> p.id + AND o.doi = regexp_replace(p.doi, '^(https?://(dx\.)?doi\.org/|doi:)', '', 'i') + );