Cited Paper Retraction Watch with Crossref, DuckDB and Slack

id: cited-paper-retraction-watch
namespace: company.team

description: |
  Watch the papers your models, documentation or policies cite and find out when one of
  them is retracted, corrected or flagged with an expression of concern. Crossref records
  every such notice against the original DOI (`updated-by`), including the ones reported
  by Retraction Watch. This flow checks the whole watchlist with one request, keeps the
  notices already seen in the KV store, fetches the notice record for each new one and
  alerts with title, date and link. It also catches DOIs that Crossref does not know, so a
  typo in a citation surfaces as NOT_FOUND instead of silently passing.

inputs:
  - id: dois
    type: STRING
    defaults: "10.1038/nature14539;10.1126/science.aac4716;10.1177/1758835919874651\
      ;10.1016/S0140-6736(97)11096-0"
    description: DOIs to watch, separated by semicolons. The defaults are two clean
      papers and two retracted ones so the first run shows both outcomes.

  - id: contact_email
    type: STRING
    defaults: ""
    description: Your email for Crossref's polite pool (faster, more reliable rate
      limits). Leave empty to use the anonymous pool.

  - id: alert_on_corrections
    type: BOOL
    defaults: false
    description: Alert on corrections and errata too. By default only retractions,
      partial retractions, expressions of concern, removals and withdrawals
      alert; corrections are logged.

triggers:
  - id: weekly_check
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 6 * * 2"
    description: Crossref receives a few hundred retraction notices a week; a weekly
      check is enough for a citation list.

tasks:
  - id: normalise_dois
    type: io.kestra.plugin.core.debug.Return
    description: Strips spaces so `a; b; c` and `a;b;c` build the same filter.
    format: "{{ inputs.dois | replace({' ': ''}) | trim }}"

  - id: fetch_works
    type: io.kestra.plugin.core.http.Download
    description: |
      One request for the whole watchlist: the `doi:` filter can be repeated, and `select` keeps the
      payload to the fields needed (the `updated-by` array is where retraction and correction notices live).
    uri: "https://api.crossref.org/works?filter=doi:{{ outputs.normalise_dois.value
      | split(';') | join(',doi:')
      }}&select=DOI,title,container-title,type,issued,is-referenced-by-count,up\
      dated-by&rows=1000{{ inputs.contact_email is not empty ? '&mailto=' ~
      inputs.contact_email : '' }}"
    headers:
      User-Agent: "kestra-cited-paper-retraction-watch/1.0 (https://kestra.io{{
        inputs.contact_email is not empty ? '; mailto:' ~ inputs.contact_email :
        '' }})"
      Accept: application/json

  - id: previous_notices
    type: io.kestra.plugin.core.kv.Get
    key: cited_paper_notices_seen
    errorOnMissing: false
    description: JSON array of the (doi, notice_doi, update_type) triples already
      reported. Missing on the first run, so every existing notice is reported
      once.

  - id: classify
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: |
      DuckDB parses the Crossref response, flattens each work's `updated-by` notices into one row per
      notice (a notice reported by both the publisher and Retraction Watch is one row), joins the
      watchlist to find DOIs Crossref does not know, and anti-joins against the notices already seen.
      The result row carries counts, the per-DOI status report and the new notices as JSON for the alert loop.
    url: "jdbc:duckdb:"
    fetchType: FETCH_ONE
    inputFiles:
      works.json: "{{ outputs.fetch_works.uri }}"
      seen.json: "{{ outputs.previous_notices.value ?? '[]' }}"
    outputFiles:
      - status_csv
    parameters:
      watchlist: "{{ outputs.normalise_dois.value }}"
      alert_on_corrections: "{{ inputs.alert_on_corrections }}"
    sql: |
      CREATE TABLE watchlist AS
      SELECT lower(trim(d)) AS doi FROM unnest(string_split(:watchlist, ';')) AS t(d) WHERE trim(d) <> '';

      CREATE TABLE works AS
      SELECT lower(i->>'$.DOI') AS doi,
             i->>'$.title[0]' AS title,
             i->>'$."container-title"[0]' AS venue,
             i->>'$.type' AS work_type,
             CAST(i->>'$.issued."date-parts"[0][0]' AS INTEGER) AS issued_year,
             CAST(i->>'$."is-referenced-by-count"' AS INTEGER) AS cited_by,
             coalesce(json_extract(i, '$."updated-by"'), CAST('[]' AS JSON)) AS updates
      FROM (SELECT unnest(from_json(json_extract(content, '$.message.items'), '["JSON"]')) AS i FROM read_text('works.json'));

      CREATE TABLE notices AS
      SELECT w.doi, w.title, w.venue, w.issued_year, w.cited_by,
             u->>'$.type' AS update_type,
             lower(u->>'$.DOI') AS notice_doi,
             min(CAST(u->>'$.updated."date-time"' AS DATE)) AS notice_date,
             string_agg(DISTINCT u->>'$.source', '+') AS reported_by
      FROM works w, unnest(from_json(w.updates, '["JSON"]')) AS t(u)
      GROUP BY ALL;

      CREATE TABLE seen AS
      SELECT lower(s->>'$.doi') AS doi, lower(s->>'$.notice_doi') AS notice_doi, s->>'$.update_type' AS update_type
      FROM (SELECT unnest(from_json(content, '["JSON"]')) AS s FROM read_text('seen.json'));

      CREATE TABLE new_notices AS
      SELECT n.*,
             n.update_type IN ('retraction', 'partial_retraction', 'expression_of_concern', 'removal', 'withdrawal')
               OR (:alert_on_corrections AND n.update_type IN ('correction', 'erratum', 'corrigendum')) AS alert
      FROM notices n
      ANTI JOIN seen s ON s.doi = n.doi AND s.notice_doi = n.notice_doi AND s.update_type = n.update_type;

      CREATE TABLE status AS
      SELECT wl.doi,
             w.title, w.venue, w.issued_year, w.cited_by,
             CASE WHEN w.doi IS NULL THEN 'NOT_FOUND'
                  WHEN bool_or(n.update_type IN ('retraction', 'removal', 'withdrawal')) THEN 'RETRACTED'
                  WHEN bool_or(n.update_type = 'partial_retraction') THEN 'PARTIALLY_RETRACTED'
                  WHEN bool_or(n.update_type = 'expression_of_concern') THEN 'EXPRESSION_OF_CONCERN'
                  WHEN count(n.notice_doi) > 0 THEN 'CORRECTED'
                  ELSE 'CLEAN' END AS status,
             count(n.notice_doi) AS notices,
             max(n.notice_date) AS latest_notice_date,
             string_agg(n.update_type || ' ' || n.notice_doi, '; ' ORDER BY n.notice_date) AS notice_list
      FROM watchlist wl
      LEFT JOIN works w ON w.doi = wl.doi
      LEFT JOIN notices n ON n.doi = wl.doi
      GROUP BY wl.doi, w.doi, w.title, w.venue, w.issued_year, w.cited_by;

      COPY (SELECT * FROM status ORDER BY status, doi) TO '{{ outputFiles.status_csv }}' (HEADER, DELIMITER ',');

      SELECT
        (SELECT count(*) FROM watchlist) AS watched,
        (SELECT count(*) FROM works) AS found,
        (SELECT count(*) FILTER (WHERE status = 'NOT_FOUND') FROM status) AS not_found,
        (SELECT count(*) FILTER (WHERE status IN ('RETRACTED', 'PARTIALLY_RETRACTED')) FROM status) AS retracted,
        (SELECT count(*) FILTER (WHERE status = 'EXPRESSION_OF_CONCERN') FROM status) AS concerns,
        (SELECT count(*) FILTER (WHERE status = 'CORRECTED') FROM status) AS corrected,
        (SELECT count(*) FILTER (WHERE status = 'CLEAN') FROM status) AS clean,
        (SELECT count(*) FROM seen) = 0 AS is_baseline,
        (SELECT count(*) FROM new_notices) AS new_notices,
        (SELECT count(*) FILTER (WHERE alert) FROM new_notices) AS new_alerts,
        (SELECT coalesce(string_agg(doi, '; '), '') FROM status WHERE status = 'NOT_FOUND') AS not_found_dois,
        (SELECT coalesce(CAST(json_group_array(json_object('doi', doi, 'title', title, 'venue', venue, 'issued_year', issued_year, 'cited_by', cited_by, 'update_type', update_type, 'notice_doi', notice_doi, 'notice_date', notice_date, 'reported_by', reported_by)) AS VARCHAR), '[]')
           FROM new_notices WHERE alert) AS alert_notices,
        (SELECT coalesce(CAST(json_group_array(json_object('doi', doi, 'notice_doi', notice_doi, 'update_type', update_type)) AS VARCHAR), '[]')
           FROM (SELECT doi, notice_doi, update_type FROM notices UNION SELECT doi, notice_doi, update_type FROM seen)) AS seen_now;

  - id: save_notices
    type: io.kestra.plugin.core.kv.Set
    description: Everything seen so far, so a notice is reported once. Stored as a
      JSON array of triples; a few kilobytes for hundreds of papers.
    key: cited_paper_notices_seen
    value: "{{ outputs.classify.outputs[0].row.seen_now }}"

  - id: report_new_notices
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.classify.outputs[0].row.new_alerts > 0 }}"
    then:
      - id: each_notice
        type: io.kestra.plugin.core.flow.Loop
        description: One alert per new notice, with the notice record fetched from
          Crossref so the message carries the notice title, date and link.
        values: "{{ outputs.classify.outputs[0].row.alert_notices }}"
        concurrencyLimit: 2
        tasks:
          - id: fetch_notice
            type: io.kestra.plugin.core.http.Request
            uri: "https://api.crossref.org/works/{{ fromJson(item.value).notice_doi |
              urlencode }}{{ inputs.contact_email is not empty ? '?mailto=' ~
              inputs.contact_email : '' }}"
            headers:
              User-Agent: "kestra-cited-paper-retraction-watch/1.0 (https://kestra.io{{
                inputs.contact_email is not empty ? '; mailto:' ~
                inputs.contact_email : '' }})"
              Accept: application/json
            allowFailure: true
            description: Some notices are registered with Crossref by Retraction Watch only;
              the lookup can 404 and the alert still goes out with the DOI link.

          - id: notify
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            allowFailure: true
            description: allowFailure keeps the check green when the webhook is not
              configured; the notice is still in the status report and the logs.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": ":rotating_light: *{{ fromJson(item.value).update_type | upper | replace({'_': ' '}) }}* of a cited paper\n*{{ fromJson(item.value).title }}* ({{ fromJson(item.value).venue }}, {{ fromJson(item.value).issued_year }}, cited {{ fromJson(item.value).cited_by }} times)\nhttps://doi.org/{{ fromJson(item.value).doi }}\nNotice ({{ fromJson(item.value).notice_date }}, reported by {{ fromJson(item.value).reported_by }}): {{ fromJson(outputs.fetch_notice.body).message.title[0] ?? 'see notice' }}\nhttps://doi.org/{{ fromJson(item.value).notice_doi }}\nExecution: {{ kestra.url ?? '' }}/ui/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}"
              }

          - id: log_notice
            type: io.kestra.plugin.core.log.Log
            level: WARN
            message: "{{ fromJson(item.value).update_type | upper }}: {{
              fromJson(item.value).doi }} ({{ fromJson(item.value).title }}) by
              notice {{ fromJson(item.value).notice_doi }} on {{
              fromJson(item.value).notice_date }}: {{
              fromJson(outputs.fetch_notice.body).message.title[0] ?? 'notice
              record not in Crossref' }}"
    else:
      - id: nothing_new
        type: io.kestra.plugin.core.log.Log
        message: "No new retraction or concern notices for {{
          outputs.classify.outputs[0].row.watched }} watched DOIs ({{
          outputs.classify.outputs[0].row.retracted }} already retracted, {{
          outputs.classify.outputs[0].row.corrected }} corrected, {{
          outputs.classify.outputs[0].row.clean }} clean, {{
          outputs.classify.outputs[0].row.not_found }} not found)."

  - id: warn_not_found
    type: io.kestra.plugin.core.log.Log
    runIf: "{{ outputs.classify.outputs[0].row.not_found > 0 }}"
    level: WARN
    message: "Crossref does not know these DOIs (typo, or not registered with
      Crossref): {{ outputs.classify.outputs[0].row.not_found_dois }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: |
      Watched {{ outputs.classify.outputs[0].row.watched }} DOIs, found {{ outputs.classify.outputs[0].row.found }}{{ outputs.classify.outputs[0].row.is_baseline ? ' (BASELINE run)' : '' }}.
      Retracted {{ outputs.classify.outputs[0].row.retracted }}, expressions of concern {{ outputs.classify.outputs[0].row.concerns }}, corrected {{ outputs.classify.outputs[0].row.corrected }}, clean {{ outputs.classify.outputs[0].row.clean }}, not found {{ outputs.classify.outputs[0].row.not_found }}.
      New notices this run: {{ outputs.classify.outputs[0].row.new_notices }} ({{ outputs.classify.outputs[0].row.new_alerts }} alerted).
      Status report: {{ outputs.classify.outputFiles.status_csv }}

outputs:
  - id: summary
    type: JSON
    value: "{{ outputs.classify.outputs[0].row }}"
  - id: status_csv
    type: FILE
    value: "{{ outputs.classify.outputFiles.status_csv }}"

Model cards, clinical guidelines and policy briefs cite papers that nobody reads again after publication, yet Crossref receives a few hundred retraction notices a week and the citation keeps pointing at a result that no longer stands. Crossref records every retraction, partial retraction, expression of concern and correction against the original DOI in the updated-by array of the work record, including the notices from the Retraction Watch database. This flow turns that array into a weekly check: one request for the whole watchlist, a DuckDB pass that flattens and classifies the notices, a KV-backed memory of what was already reported, and one alert per new notice with the notice title, date and link.

How it works

A weekly Schedule trigger (Tuesday 06:00 UTC) runs the check; dois is a semicolon-separated list and normalise_dois (Return) strips spaces so hand-edited lists work.
fetch_works (HTTP Download) calls api.crossref.org/works once with a repeated filter=doi: for every DOI and select limited to the fields needed, including updated-by. If contact_email is set it is appended as mailto= and put in the User-Agent, which moves the request to Crossref's polite pool.
previous_notices (KV Get, errorOnMissing: false) reads the JSON array of (doi, notice_doi, update_type) triples reported by earlier runs; it is missing on the first run.
classify (DuckDB Queries, fetchType: FETCH_ONE) parses the response with read_text and from_json, flattens each work's updated-by into one row per notice (a notice reported by both the publisher and Retraction Watch is one row, reported_by keeps both sources), left-joins the watchlist so a DOI Crossref does not know becomes NOT_FOUND, and anti-joins the notices against the ones already seen. Each DOI gets a status: RETRACTED, PARTIALLY_RETRACTED, EXPRESSION_OF_CONCERN, CORRECTED, CLEAN or NOT_FOUND. The per-DOI report is written as CSV through outputFiles; the single result row carries the counts, the new notices to alert on as JSON and the merged seen set.
save_notices (KV Set) stores the merged set, so each notice is reported exactly once and DOIs removed from the list do not lose their history.
report_new_notices (If) runs each_notice (Loop) over the new alert notices when there are any. Each iteration calls fetch_notice (HTTP Request) for the notice record so the message has the notice title, then posts to Slack (allowFailure, so a missing webhook warns instead of failing) and logs a WARN line. Retractions, partial retractions, expressions of concern, removals and withdrawals always alert; corrections and errata alert only when alert_on_corrections is true, otherwise they are counted and kept in the report.
warn_not_found logs the DOIs Crossref does not know (runIf on the count), which catches typos in a citation list.
summary logs the counts and the flow exposes the summary JSON and the status CSV as outputs.

What you get

A CSV per run with one row per watched DOI: title, venue, year, citation count, status, number of notices, latest notice date and the list of notice DOIs.
One alert per new notice, never repeated, with the paper, the notice type, the notice date, who reported it (publisher, Retraction Watch or both) and doi.org links to both records.
NOT_FOUND for DOIs Crossref has no record of, so a mistyped citation fails loudly instead of passing as clean.
State in the KV store only: no database, no files to manage, a few kilobytes for hundreds of papers.

Who it's for

ML and data science teams that cite papers in model cards and want to know if a method they rely on was retracted.
Medical writers, guideline committees and systematic reviewers who maintain a reference list over years.
Research software engineers and librarians who run integrity checks on institutional publication lists.

Why orchestrate this with Kestra

Download plus DuckDB inputFiles turns a JSON API response into tables without a script; the classification is SQL you can read in the flow.
The KV store holds the seen set between runs, so "report once" needs no database.
Loop runs one alert per notice with its own HTTP lookup, and allowFailure keeps a missing webhook or notice record from failing the check.
The watchlist is an input: the same flow serves one model card or a whole lab's publication list.

Prerequisites

Kestra 2.0 with the JDBC DuckDB plugin and, for the alert, the Slack plugin (io.kestra.plugin.slack.notifications.SlackIncomingWebhook).
Outbound HTTPS to api.crossref.org; the REST API needs no key. Set contact_email for the polite pool on larger lists.
DOIs must be registered with Crossref (most journals and conference proceedings are; DataCite DOIs for datasets are not and show as NOT_FOUND).

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook for the alert (optional; without it the notify task warns and the check still succeeds).

Quick start

Import the flow and click Execute with the defaults. The list holds two clean papers and two retracted ones (one 2019 oncology paper retracted in 2021, and the 1998 Lancet paper retracted in 2010 after a 2004 partial retraction). The first run is a baseline: it reports retracted 2, clean 2, finds 3 notices and alerts on the 2 retractions.
Execute again: new_notices 0, the nothing_new branch runs.
Add a DOI with a typo to the list and run: it shows as NOT_FOUND in the CSV and warn_not_found names it.
Set alert_on_corrections to true after deleting the KV key cited_paper_notices_seen: the 2004 correction notice now alerts as well.
Replace the defaults with your own DOIs, add SLACK_WEBHOOK_URL as a secret, set contact_email and leave the weekly schedule on.

How to extend

Read the DOI list from a namespace file or a Google Sheet instead of an input, so authors edit the list without touching the flow.
Open a Jira or GitHub issue per notice with the Jira or GitHub plugins in place of, or in addition to, the Slack message.
Append each run's status CSV to a warehouse table to keep a history of when each paper changed status.
Lower the schedule to daily for lists under active review; one request per run keeps it well inside Crossref's limits.

Watch Cited Papers for Retractions, Expressions of Concern and Corrections via Crossref

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Watch Cited Papers for Retractions, Expressions of Concern and Corrections via Crossref

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Citus Inactive Worker Reactivate Guard

Incremental Docs-to-Chroma RAG Sync with Deletions and a Retrieval Smoke Test

Score change requests on risk and build the CAB agenda, pre-approving the low-risk ones

Chroma Collection Distance Metric Validator

Citus Colocation Violation Guard

Citus Reference Table Replica Guard

New to Kestra?