fix(drug-reference): fold dataset freshness into the content-auto-update path
The drug dataset auto-updated on its own daily cron that ignored the contentAutoUpdate.* master switch, so it would refresh even with content auto-update turned off, and it didn't ride the content-update path the way ZIMs and maps do. Move the export_date freshness/apply orchestration onto DrugReferenceService.attemptAutoUpdate(), add ContentAutoUpdateService.attemptDrugDataset() gated on the same enabled + window config, and have the hourly ContentAutoUpdateJob drive both. Retire the standalone DrugAutoUpdateJob. The ZIM/map attempt() path is unchanged. Addresses the upstream #1040 request to wire the openFDA export_date check into the content updater so the dataset updates alongside ZIMs and maps. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Chris
parent
22eda3040e
commit
a6d94edbee
|
|
@ -30,7 +30,17 @@ export class ContentAutoUpdateJob {
|
|||
|
||||
const result = await contentAutoUpdateService.attempt()
|
||||
logger.info(`[ContentAutoUpdateJob] ${result.started} started: ${result.reason}`)
|
||||
return result
|
||||
|
||||
// The FDA drug dataset refreshes on the same cycle and master switch as the
|
||||
// ZIM/map catalog (its apply path differs, so it's a separate step, not part
|
||||
// of attempt()). Governed by the same `contentAutoUpdate.*` settings.
|
||||
const drugResult = await contentAutoUpdateService.attemptDrugDataset()
|
||||
logger.info(`[ContentAutoUpdateJob] ${drugResult.started} started: ${drugResult.reason}`)
|
||||
|
||||
return {
|
||||
started: result.started + drugResult.started,
|
||||
reason: `${result.reason}; ${drugResult.reason}`,
|
||||
}
|
||||
}
|
||||
|
||||
static async schedule() {
|
||||
|
|
|
|||
|
|
@ -300,8 +300,8 @@ export class DownloadDrugDataJob {
|
|||
/**
|
||||
* Fetch + parse the openFDA download manifest. The SINGLE source of truth for
|
||||
* the openFDA manifest call (Maxim 4): the download job uses it on pass 0, and
|
||||
* the freshness check (DrugReferenceService.checkForUpdate → DrugAutoUpdateJob)
|
||||
* reuses it so there is exactly one place that knows the URL and the
|
||||
* the freshness check (DrugReferenceService.checkForUpdate, driven by
|
||||
* attemptAutoUpdate) reuses it so there is exactly one place that knows the URL and the
|
||||
* offline-error translation.
|
||||
*/
|
||||
static async fetchManifest(): Promise<DrugLabelManifest> {
|
||||
|
|
|
|||
|
|
@ -1,137 +0,0 @@
|
|||
import { Job } from 'bullmq'
|
||||
import { QueueService } from '#services/queue_service'
|
||||
import { DrugReferenceService } from '#services/drug_reference_service'
|
||||
import { DownloadDrugDataJob } from '#jobs/download_drug_data_job'
|
||||
import { IngestDrugDataJob } from '#jobs/ingest_drug_data_job'
|
||||
import logger from '@adonisjs/core/services/logger'
|
||||
|
||||
/**
|
||||
* Daily job that checks the openFDA manifest's `export_date` against the
|
||||
* last-ingested one and, when newer, re-downloads the FDA drug dataset (the
|
||||
* download auto-chains the ingest, which advances the stored export_date).
|
||||
*
|
||||
* Mirrors {@link ContentAutoUpdateJob} (the ZIM/map freshness loop) but stays a
|
||||
* SEPARATE job rather than extending it: the drug dataset isn't an
|
||||
* `InstalledResource` catalog row with a filename `YYYY-MM` version, so its
|
||||
* freshness key (export_date) and apply path (the drug download/ingest chain)
|
||||
* don't fit the ZIM loop. openFDA publishes ~weekly; a daily check is safe and
|
||||
* bandwidth-respectful.
|
||||
*
|
||||
* Gating (kept deliberately simple for v1):
|
||||
* - Only acts when the dataset is actually INSTALLED (rows present). An update
|
||||
* for something not installed is meaningless.
|
||||
* - Skips when a drug download OR ingest is already in flight, so it never
|
||||
* stacks a refresh on top of a running install/update.
|
||||
* The 1.7 GB transfer respects those guards; a window/cap analogous to the ZIM
|
||||
* `contentAutoUpdate.*` keys is a follow-up (open question for the maintainer:
|
||||
* share `contentAutoUpdate.*` or get `drugReference.autoUpdate.*`).
|
||||
*/
|
||||
export class DrugAutoUpdateJob {
|
||||
static get queue() {
|
||||
return 'system'
|
||||
}
|
||||
|
||||
static get key() {
|
||||
return 'drug-auto-update'
|
||||
}
|
||||
|
||||
async handle(_job: Job) {
|
||||
logger.info('[DrugAutoUpdateJob] Evaluating drug-reference freshness...')
|
||||
|
||||
const service = new DrugReferenceService()
|
||||
|
||||
// Only act when the dataset is installed.
|
||||
const rowCount = await service.rowCount()
|
||||
if (rowCount === 0) {
|
||||
logger.info('[DrugAutoUpdateJob] Drug dataset not installed — skipping.')
|
||||
return { started: false, reason: 'not-installed' }
|
||||
}
|
||||
|
||||
// Don't stack on top of an in-flight download/ingest.
|
||||
const [dlJob, ingJob] = await Promise.all([
|
||||
DownloadDrugDataJob.getJob(),
|
||||
IngestDrugDataJob.getJob(),
|
||||
])
|
||||
const activeStates = ['active', 'waiting', 'delayed']
|
||||
const dlState = dlJob ? await dlJob.getState() : undefined
|
||||
const ingState = ingJob ? await ingJob.getState() : undefined
|
||||
if (
|
||||
(dlState && activeStates.includes(dlState)) ||
|
||||
(ingState && activeStates.includes(ingState))
|
||||
) {
|
||||
logger.info('[DrugAutoUpdateJob] A drug job is already in flight — skipping.')
|
||||
return { started: false, reason: 'job-in-flight' }
|
||||
}
|
||||
|
||||
let check: Awaited<ReturnType<DrugReferenceService['checkForUpdate']>>
|
||||
try {
|
||||
check = await service.checkForUpdate()
|
||||
} catch (err) {
|
||||
// Offline or manifest fetch failed — a transient miss, not an error worth
|
||||
// failing the scheduled job over. Try again next day.
|
||||
logger.warn(
|
||||
`[DrugAutoUpdateJob] Freshness check failed (will retry next run): ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}`
|
||||
)
|
||||
return { started: false, reason: 'check-failed' }
|
||||
}
|
||||
|
||||
if (!check.updateAvailable) {
|
||||
logger.info(
|
||||
`[DrugAutoUpdateJob] Up to date (current=${check.currentExportDate ?? 'none'}, ` +
|
||||
`latest=${check.latestExportDate ?? 'unknown'}).`
|
||||
)
|
||||
return { started: false, reason: 'up-to-date' }
|
||||
}
|
||||
|
||||
logger.info(
|
||||
`[DrugAutoUpdateJob] Newer export_date available ` +
|
||||
`(current=${check.currentExportDate ?? 'none'} → latest=${check.latestExportDate}); ` +
|
||||
'triggering re-download.'
|
||||
)
|
||||
const result = await service.triggerDownload()
|
||||
return {
|
||||
started: result.created,
|
||||
reason: result.created ? 'update-dispatched' : result.message,
|
||||
latestExportDate: check.latestExportDate,
|
||||
}
|
||||
}
|
||||
|
||||
static async schedule() {
|
||||
const queueService = QueueService.getInstance()
|
||||
const queue = queueService.getQueue(this.queue)
|
||||
|
||||
await queue.upsertJobScheduler(
|
||||
'daily-drug-auto-update',
|
||||
{ pattern: '30 3 * * *' }, // 03:30 daily — off-peak, distinct from the hourly content loop
|
||||
{
|
||||
name: this.key,
|
||||
opts: {
|
||||
removeOnComplete: { count: 7 },
|
||||
removeOnFail: { count: 5 },
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
logger.info('[DrugAutoUpdateJob] Drug auto-update evaluation scheduled with cron: 30 3 * * *')
|
||||
}
|
||||
|
||||
static async dispatch() {
|
||||
const queueService = QueueService.getInstance()
|
||||
const queue = queueService.getQueue(this.queue)
|
||||
|
||||
const job = await queue.add(
|
||||
this.key,
|
||||
{},
|
||||
{
|
||||
attempts: 1,
|
||||
removeOnComplete: { count: 7 },
|
||||
removeOnFail: { count: 5 },
|
||||
}
|
||||
)
|
||||
|
||||
logger.info(`[DrugAutoUpdateJob] Dispatched ad-hoc drug auto-update evaluation job ${job.id}`)
|
||||
return job
|
||||
}
|
||||
}
|
||||
|
|
@ -4,6 +4,7 @@ import KVStore from '#models/kv_store'
|
|||
import InstalledResource from '#models/installed_resource'
|
||||
import { DownloadService } from '#services/download_service'
|
||||
import { CollectionUpdateService } from '#services/collection_update_service'
|
||||
import { DrugReferenceService } from '#services/drug_reference_service'
|
||||
import {
|
||||
KiwixCatalogService,
|
||||
reconcileResourceUpdateState,
|
||||
|
|
@ -269,8 +270,11 @@ export class ContentAutoUpdateService {
|
|||
await this.maybeResetWindowBudget(config, now)
|
||||
|
||||
// Local catalog check + persist available-update state for every resource.
|
||||
// ZIM/map catalog path only — `dataset` resources are excluded (they get
|
||||
// their own freshness path; no-op today as none are written in this slice).
|
||||
// ZIM/map catalog path only — `dataset` resources are excluded here because
|
||||
// their freshness key (openFDA `export_date`) and apply path (the drug
|
||||
// download/ingest chain) don't fit the catalog-version model. They refresh
|
||||
// via `attemptDrugDataset()`, which the same content-update job runs under
|
||||
// the same master switch + window.
|
||||
const installed = await InstalledResource.query().whereNot('resource_type', 'dataset')
|
||||
const latestByKey = await this.catalog.getLatestForResources(
|
||||
// `dataset` rows are filtered out above, so the type narrows to ZIM/map.
|
||||
|
|
@ -379,6 +383,39 @@ export class ContentAutoUpdateService {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Freshness pass for the FDA drug dataset (`resource_type` 'dataset'), run by
|
||||
* the same hourly ContentAutoUpdateJob as the ZIM/map `attempt()` and gated on
|
||||
* the SAME master switch + window. The dataset's apply path differs from the
|
||||
* catalog loop (openFDA `export_date` + the drug download/ingest chain rather
|
||||
* than an InstalledResource catalog-version row), so it runs as its own step
|
||||
* instead of through `selectUnderCap` — but sharing `contentAutoUpdate.*` means
|
||||
* it never updates while content auto-update is off, and refreshes alongside
|
||||
* ZIMs and maps when it is on. Isolated so a drug-side failure can't affect the
|
||||
* catalog run.
|
||||
*/
|
||||
async attemptDrugDataset(): Promise<{ started: number; reason: string }> {
|
||||
const config = await this.getConfig()
|
||||
if (!config.enabled) {
|
||||
return { started: 0, reason: 'Content auto-update is disabled' }
|
||||
}
|
||||
if (!isWithinWindow(config.windowStart, config.windowEnd, DateTime.now())) {
|
||||
return {
|
||||
started: 0,
|
||||
reason: `Outside update window (${config.windowStart}-${config.windowEnd})`,
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await new DrugReferenceService().attemptAutoUpdate()
|
||||
return { started: result.started ? 1 : 0, reason: `drug dataset: ${result.reason}` }
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error)
|
||||
logger.warn(`[ContentAutoUpdateService] Drug dataset freshness check failed: ${message}`)
|
||||
return { started: 0, reason: `drug dataset check failed: ${message}` }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Evaluate what the next run *would* do, without hitting the network,
|
||||
* persisting state, or dispatching anything. Operates on the available-update
|
||||
|
|
|
|||
|
|
@ -408,15 +408,15 @@ export class DrugReferenceService {
|
|||
}
|
||||
|
||||
/**
|
||||
* Freshness check for the daily auto-update path: compare the live openFDA
|
||||
* Freshness check for the content-auto-update path: compare the live openFDA
|
||||
* manifest `export_date` against the last-ingested one (KV
|
||||
* `drugReference.lastUpdatedExportDate`). Reuses the job's single manifest
|
||||
* fetch (Maxim 4 — one openFDA call site) and the pure `isExportDateNewer`
|
||||
* compare (defensive about the unconfirmed date format; see its TODO).
|
||||
*
|
||||
* Returns `{ updateAvailable, latestExportDate, currentExportDate }`. Never
|
||||
* triggers a download itself — the caller (DrugAutoUpdateJob) decides whether
|
||||
* to, after its own gating (only when installed, no active job).
|
||||
* triggers a download itself — `attemptAutoUpdate` decides whether to, after
|
||||
* its own gating (only when installed, no active job).
|
||||
*/
|
||||
async checkForUpdate(): Promise<{
|
||||
updateAvailable: boolean
|
||||
|
|
@ -433,6 +433,79 @@ export class DrugReferenceService {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Freshness pass for the drug dataset, invoked from
|
||||
* `ContentAutoUpdateService.attemptDrugDataset()` inside the hourly content
|
||||
* loop — so the dataset refreshes alongside ZIMs and maps under the shared
|
||||
* `contentAutoUpdate.*` master switch and window, not on a rogue schedule.
|
||||
*
|
||||
* Gating (unchanged from the prior standalone job): only acts when the dataset
|
||||
* is installed, and never while a drug download or ingest is already in flight,
|
||||
* so it can't stack a refresh on a running install. A failed manifest fetch is
|
||||
* a transient miss (offline), reported rather than thrown.
|
||||
*/
|
||||
async attemptAutoUpdate(): Promise<{
|
||||
started: boolean
|
||||
reason: string
|
||||
latestExportDate?: string | null
|
||||
}> {
|
||||
// Only act when the dataset is installed.
|
||||
const rowCount = await this.rowCount()
|
||||
if (rowCount === 0) {
|
||||
return { started: false, reason: 'not-installed' }
|
||||
}
|
||||
|
||||
// Don't stack on top of an in-flight download/ingest.
|
||||
const [dlJob, ingJob] = await Promise.all([
|
||||
DownloadDrugDataJob.getJob(),
|
||||
IngestDrugDataJob.getJob(),
|
||||
])
|
||||
const activeStates = ['active', 'waiting', 'delayed']
|
||||
const dlState = dlJob ? await dlJob.getState() : undefined
|
||||
const ingState = ingJob ? await ingJob.getState() : undefined
|
||||
if (
|
||||
(dlState && activeStates.includes(dlState)) ||
|
||||
(ingState && activeStates.includes(ingState))
|
||||
) {
|
||||
return { started: false, reason: 'job-in-flight' }
|
||||
}
|
||||
|
||||
let check: Awaited<ReturnType<DrugReferenceService['checkForUpdate']>>
|
||||
try {
|
||||
check = await this.checkForUpdate()
|
||||
} catch (err) {
|
||||
// Offline or manifest fetch failed — transient, retried next run.
|
||||
logger.warn(
|
||||
`[DrugReferenceService] Freshness check failed (will retry next run): ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}`
|
||||
)
|
||||
return { started: false, reason: 'check-failed' }
|
||||
}
|
||||
|
||||
if (!check.updateAvailable) {
|
||||
return {
|
||||
started: false,
|
||||
reason: `up-to-date (current=${check.currentExportDate ?? 'none'}, latest=${
|
||||
check.latestExportDate ?? 'unknown'
|
||||
})`,
|
||||
latestExportDate: check.latestExportDate,
|
||||
}
|
||||
}
|
||||
|
||||
logger.info(
|
||||
`[DrugReferenceService] Newer export_date available ` +
|
||||
`(current=${check.currentExportDate ?? 'none'} → latest=${check.latestExportDate}); ` +
|
||||
'triggering re-download.'
|
||||
)
|
||||
const result = await this.triggerDownload()
|
||||
return {
|
||||
started: result.created,
|
||||
reason: result.created ? 'update-dispatched' : result.message,
|
||||
latestExportDate: check.latestExportDate,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Dispatch the ingest phase from the already-downloaded on-disk parts (the
|
||||
* manual "Ingest into search" path). Guards on the KV download-state marker so
|
||||
|
|
|
|||
|
|
@ -14,7 +14,6 @@ import { AppAutoUpdateJob } from '#jobs/app_auto_update_job'
|
|||
import { ContentAutoUpdateJob } from '#jobs/content_auto_update_job'
|
||||
import { DownloadDrugDataJob } from '#jobs/download_drug_data_job'
|
||||
import { IngestDrugDataJob } from '#jobs/ingest_drug_data_job'
|
||||
import { DrugAutoUpdateJob } from '#jobs/drug_auto_update_job'
|
||||
|
||||
export default class QueueWork extends BaseCommand {
|
||||
static commandName = 'queue:work'
|
||||
|
|
@ -151,7 +150,6 @@ export default class QueueWork extends BaseCommand {
|
|||
await AutoUpdateJob.schedule()
|
||||
await AppAutoUpdateJob.schedule()
|
||||
await ContentAutoUpdateJob.schedule()
|
||||
await DrugAutoUpdateJob.schedule()
|
||||
|
||||
// Safety net: log unhandled rejections instead of crashing the worker process.
|
||||
// Individual job errors are already caught by BullMQ; this catches anything that
|
||||
|
|
@ -185,7 +183,6 @@ export default class QueueWork extends BaseCommand {
|
|||
handlers.set(AutoUpdateJob.key, new AutoUpdateJob())
|
||||
handlers.set(AppAutoUpdateJob.key, new AppAutoUpdateJob())
|
||||
handlers.set(ContentAutoUpdateJob.key, new ContentAutoUpdateJob())
|
||||
handlers.set(DrugAutoUpdateJob.key, new DrugAutoUpdateJob())
|
||||
handlers.set(DownloadDrugDataJob.key, new DownloadDrugDataJob())
|
||||
handlers.set(IngestDrugDataJob.key, new IngestDrugDataJob())
|
||||
|
||||
|
|
@ -199,7 +196,6 @@ export default class QueueWork extends BaseCommand {
|
|||
queues.set(AutoUpdateJob.key, AutoUpdateJob.queue)
|
||||
queues.set(AppAutoUpdateJob.key, AppAutoUpdateJob.queue)
|
||||
queues.set(ContentAutoUpdateJob.key, ContentAutoUpdateJob.queue)
|
||||
queues.set(DrugAutoUpdateJob.key, DrugAutoUpdateJob.queue)
|
||||
queues.set(DownloadDrugDataJob.key, DownloadDrugDataJob.queue)
|
||||
queues.set(IngestDrugDataJob.key, IngestDrugDataJob.queue)
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue