feat(drug-reference): rework into an opt-in medicine-standard tier dataset
Reshapes the offline FDA drug reference from a built-in feature into curated
content installed by selecting the Medicine / Standard tier, per maintainer
direction.
- install-state: the ingest writes an installed_resources 'dataset' row on
ready (version = the openFDA export_date), threaded installer to download to
ingest; the tier-status math and the home-tile gate read it. Manual ingests
write no row, so install-state stays tied to the curated path.
- manifest: declare the dataset in the medicine-standard tier (runtime fetches
the remote manifest, so this also needs to land upstream).
- install-gating: the drug-reference home tiles render only when installed.
- uninstall: DrugReferenceService.uninstall() stops the two drug queues, deletes
the on-disk parts, truncates drug_labels (schema kept), clears the KV markers,
and drops the install row. Best-effort, logged, scoped to drug data only.
- downloads: the download phase reports the canonical {percent, downloadedBytes,
totalBytes} shape as one drug-data card in the Active Downloads aggregator with
cancel/remove; the heavy ingest stays in the IngestStatus surface with an
Indexing handoff on the card.
- auto-update: a daily DrugAutoUpdateJob compares the manifest export_date and
re-downloads when newer, gated on installed + no active job.
typecheck clean; the drug standalone suites pass. Three points are flagged in
code for the maintainer: the InstalledResource 'dataset' approach, the tier
home, and the export_date string format.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Chris
parent
754b0b1921
commit
374ccda92a
|
|
@ -224,6 +224,31 @@ export default class DrugReferenceController {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* POST /api/drug-reference/uninstall
|
||||
* Uninstall the offline FDA drug dataset: stop in-flight jobs, delete on-disk
|
||||
* parts, TRUNCATE drug_labels, clear KV markers, and remove the install-state
|
||||
* row (which auto-hides the home tiles). The curated-tier "remove" action.
|
||||
* Reports partial failures rather than masking them.
|
||||
*/
|
||||
async uninstall({ response }: HttpContext) {
|
||||
try {
|
||||
const result = await this.service.uninstall()
|
||||
if (!result.success) {
|
||||
return response.internalServerError({
|
||||
success: false,
|
||||
rowsDropped: result.rowsDropped,
|
||||
error: result.message,
|
||||
})
|
||||
}
|
||||
return { success: true, rowsDropped: result.rowsDropped, message: result.message }
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err)
|
||||
logger.error(`[DrugReferenceController] uninstall failed: ${msg}`)
|
||||
return response.internalServerError({ error: 'Could not uninstall drug reference' })
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* GET /api/drug-reference/ingest-log
|
||||
* Tail the persisted app log for ingest/download lines. In production the logger
|
||||
|
|
|
|||
|
|
@ -1,4 +1,6 @@
|
|||
import { SystemService } from '#services/system_service'
|
||||
import { DrugReferenceService } from '#services/drug_reference_service'
|
||||
import logger from '@adonisjs/core/services/logger'
|
||||
import { inject } from '@adonisjs/core'
|
||||
import type { HttpContext } from '@adonisjs/core/http'
|
||||
|
||||
|
|
@ -18,7 +20,38 @@ export default class HomeController {
|
|||
return inertia.render('home', {
|
||||
system: {
|
||||
services
|
||||
}
|
||||
},
|
||||
// Gate the Drug Reference / "When to use what" tiles behind the FDA
|
||||
// dataset install state. Installed when the curated-tier ingest has
|
||||
// reached 'ready' OR an install is in flight (downloading/ingesting) —
|
||||
// so the tile appears the moment the user opts in and persists through
|
||||
// the long install, rather than popping in only at the very end.
|
||||
drugReferenceInstalled: await this.computeDrugReferenceInstalled(),
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* True when the offline FDA drug dataset is installed or installing. Reads the
|
||||
* two-phase ingest status: ready (fully installed) or an active phase
|
||||
* (downloading/downloaded/ingesting). rowCount > 0 covers a populated table
|
||||
* whose job history was pruned. Never throws — a status read failure hides the
|
||||
* tiles (fail-closed) rather than 500-ing the dashboard.
|
||||
*/
|
||||
private async computeDrugReferenceInstalled(): Promise<boolean> {
|
||||
try {
|
||||
const status = await new DrugReferenceService().getIngestStatus()
|
||||
const installing =
|
||||
status.phase === 'downloading' ||
|
||||
status.phase === 'downloaded' ||
|
||||
status.phase === 'ingesting'
|
||||
return status.phase === 'ready' || installing || status.rowCount > 0
|
||||
} catch (err) {
|
||||
logger.error(
|
||||
`[HomeController] drug-reference install check failed: ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}`
|
||||
)
|
||||
return false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -3,11 +3,12 @@ import { access, mkdir, constants } from 'node:fs/promises'
|
|||
import logger from '@adonisjs/core/services/logger'
|
||||
import { QueueService } from '#services/queue_service'
|
||||
import { doResumableDownload } from '../utils/downloads.js'
|
||||
import { parseDrugLabelManifest, partZipPath } from '../../util/drug_labels.js'
|
||||
import { parseDrugLabelManifest, partZipPath, manifestBytesTotal } from '../../util/drug_labels.js'
|
||||
import type {
|
||||
DownloadDrugDataJobParams,
|
||||
DrugLabelManifest,
|
||||
DownloadStateMarker,
|
||||
DrugDatasetResourceMeta,
|
||||
} from '../../types/drug_reference.js'
|
||||
|
||||
/** Where all part zips are staged on the bind-mounted storage volume. */
|
||||
|
|
@ -49,8 +50,11 @@ export class DownloadDrugDataJob {
|
|||
* file level by doResumableDownload).
|
||||
*
|
||||
* @param autoChain - dispatch the ingest phase after the last part. Default true.
|
||||
* @param resourceMeta - install-state identity when the install came through a
|
||||
* curated tier. Carried through the chain so the ingest job writes the
|
||||
* `installed_resources` row on `ready`. Omit for a manual download (no row).
|
||||
*/
|
||||
static async dispatch(autoChain = true) {
|
||||
static async dispatch(autoChain = true, resourceMeta?: DrugDatasetResourceMeta) {
|
||||
const queueService = QueueService.getInstance()
|
||||
const queue = queueService.getQueue(this.queue)
|
||||
|
||||
|
|
@ -70,7 +74,7 @@ export class DownloadDrugDataJob {
|
|||
try {
|
||||
const job = await queue.add(
|
||||
this.key,
|
||||
{ autoChain } satisfies DownloadDrugDataJobParams,
|
||||
{ autoChain, resourceMeta } satisfies DownloadDrugDataJobParams,
|
||||
{
|
||||
jobId: this.jobId,
|
||||
attempts: 3,
|
||||
|
|
@ -103,6 +107,7 @@ export class DownloadDrugDataJob {
|
|||
const partIndex = params.partIndex ?? 0
|
||||
const autoChain = params.autoChain ?? true
|
||||
const startedAt = params.startedAt ?? Date.now()
|
||||
const resourceMeta = params.resourceMeta
|
||||
|
||||
logger.info(`[DownloadDrugDataJob] Starting pass partIndex=${partIndex}`)
|
||||
|
||||
|
|
@ -165,6 +170,17 @@ export class DownloadDrugDataJob {
|
|||
|
||||
await mkdir(STORAGE_BASE, { recursive: true })
|
||||
|
||||
// Aggregate-across-parts byte accounting for the Active Downloads card. The
|
||||
// drug job fans the manifest's N partitions into BullMQ continuations, but
|
||||
// the user sees ONE download — so progress is reported as bytes across the
|
||||
// whole set, not the current part. `priorPartsBytes` is the actual bytes of
|
||||
// the parts already on disk (from the running recordedParts list);
|
||||
// `manifestTotalBytes` is the sum of every partition's manifest size_mb.
|
||||
const priorRecorded =
|
||||
(params as { recordedParts?: DownloadStateMarker['parts'] }).recordedParts ?? []
|
||||
const priorPartsBytes = priorRecorded.reduce((acc, p) => acc + (p.bytes || 0), 0)
|
||||
const manifestTotalBytes = manifestBytesTotal(manifest)
|
||||
|
||||
logger.info(`[DownloadDrugDataJob] ${partition.file} → ${zipPath}`)
|
||||
let partBytes = 0
|
||||
await doResumableDownload({
|
||||
|
|
@ -176,8 +192,21 @@ export class DownloadDrugDataJob {
|
|||
partBytes = progress.downloadedBytes
|
||||
const downloadFraction = progress.downloadedBytes / (progress.totalBytes || 1)
|
||||
const pct = Math.floor(((partIndex + downloadFraction) / totalParts) * 100)
|
||||
// Fire-and-forget; swallow transient reject so it can't crash the worker.
|
||||
void job.updateProgress(pct).catch(() => {})
|
||||
const downloadedBytes = priorPartsBytes + progress.downloadedBytes
|
||||
// Emit the canonical {percent, downloadedBytes, totalBytes} object the
|
||||
// Active Downloads aggregator + the byte/speed readout expect (the old
|
||||
// bare-int progress couldn't carry bytes). parseProgress in
|
||||
// DownloadService still tolerates a bare int for back-compat, but only
|
||||
// the object surfaces a live byte/speed readout. Fire-and-forget; swallow
|
||||
// transient reject so it can't crash the worker.
|
||||
void job
|
||||
.updateProgress({
|
||||
percent: pct,
|
||||
downloadedBytes,
|
||||
totalBytes: manifestTotalBytes,
|
||||
lastProgressTime: Date.now(),
|
||||
})
|
||||
.catch(() => {})
|
||||
void job.updateData({ ...job.data, bytesDownloaded: progress.downloadedBytes }).catch(() => {})
|
||||
},
|
||||
})
|
||||
|
|
@ -204,6 +233,7 @@ export class DownloadDrugDataJob {
|
|||
autoChain,
|
||||
startedAt,
|
||||
recordedParts,
|
||||
resourceMeta,
|
||||
}
|
||||
|
||||
await queue.add(DownloadDrugDataJob.key, continuationParams, {
|
||||
|
|
@ -234,7 +264,9 @@ export class DownloadDrugDataJob {
|
|||
|
||||
if (autoChain) {
|
||||
const { IngestDrugDataJob } = await import('#jobs/ingest_drug_data_job')
|
||||
await IngestDrugDataJob.dispatch()
|
||||
// Forward the install-state identity so the ingest job writes the
|
||||
// `installed_resources` row on `ready`. undefined for a manual download.
|
||||
await IngestDrugDataJob.dispatch(resourceMeta)
|
||||
logger.info('[DownloadDrugDataJob] Auto-chained ingest phase')
|
||||
}
|
||||
}
|
||||
|
|
@ -262,6 +294,17 @@ export class DownloadDrugDataJob {
|
|||
}
|
||||
|
||||
private async fetchManifest(): Promise<DrugLabelManifest> {
|
||||
return DownloadDrugDataJob.fetchManifest()
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
* offline-error translation.
|
||||
*/
|
||||
static async fetchManifest(): Promise<DrugLabelManifest> {
|
||||
let json: unknown
|
||||
try {
|
||||
const resp = await fetch(MANIFEST_URL)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,137 @@
|
|||
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
|
||||
}
|
||||
}
|
||||
|
|
@ -10,6 +10,7 @@ import type {
|
|||
IngestDrugDataJobParams,
|
||||
DrugLabelManifest,
|
||||
DrugLabelPartition,
|
||||
DrugDatasetResourceMeta,
|
||||
} from '../../types/drug_reference.js'
|
||||
|
||||
const BATCH_SIZE = 500
|
||||
|
|
@ -157,8 +158,12 @@ export class IngestDrugDataJob {
|
|||
* Dispatch the initial ingest (pass 0). Idempotent on the deterministic jobId.
|
||||
* A finished/failed prior job under that id is cleared first so a re-ingest can
|
||||
* always restart (upserts are idempotent on set_id).
|
||||
*
|
||||
* @param resourceMeta - install-state identity, present only when the install
|
||||
* came through a curated tier. Carried through the chain so the final pass
|
||||
* writes the `installed_resources` row on `ready`. Absent on a manual ingest.
|
||||
*/
|
||||
static async dispatch() {
|
||||
static async dispatch(resourceMeta?: DrugDatasetResourceMeta) {
|
||||
const queueService = QueueService.getInstance()
|
||||
const queue = queueService.getQueue(this.queue)
|
||||
|
||||
|
|
@ -178,7 +183,7 @@ export class IngestDrugDataJob {
|
|||
try {
|
||||
const job = await queue.add(
|
||||
this.key,
|
||||
{} satisfies IngestDrugDataJobParams,
|
||||
{ resourceMeta } satisfies IngestDrugDataJobParams,
|
||||
{
|
||||
jobId: this.jobId,
|
||||
attempts: 3,
|
||||
|
|
@ -212,6 +217,7 @@ export class IngestDrugDataJob {
|
|||
const runningIngested = params.recordsIngested ?? 0
|
||||
const runningSkipped = params.recordsSkipped ?? 0
|
||||
const startedAt = params.startedAt ?? Date.now()
|
||||
const resourceMeta = params.resourceMeta
|
||||
|
||||
// Progress baseline (pass 0 only): the table's row count BEFORE this run.
|
||||
// Without it, a re-ingest into a populated table shows ~100% from second
|
||||
|
|
@ -308,6 +314,7 @@ export class IngestDrugDataJob {
|
|||
recordsSkipped: totalSkipped,
|
||||
startedAt,
|
||||
startRowCount,
|
||||
resourceMeta,
|
||||
}
|
||||
|
||||
await queue.add(IngestDrugDataJob.key, continuationParams, {
|
||||
|
|
@ -328,7 +335,7 @@ export class IngestDrugDataJob {
|
|||
})
|
||||
} else {
|
||||
// Final part — write KV status, mark ready, THEN reclaim disk.
|
||||
await this.writeFinalStatus(exportDate)
|
||||
await this.writeFinalStatus(exportDate, resourceMeta)
|
||||
|
||||
await job.updateData({
|
||||
...job.data,
|
||||
|
|
@ -731,10 +738,74 @@ export class IngestDrugDataJob {
|
|||
})
|
||||
}
|
||||
|
||||
private async writeFinalStatus(exportDate: string): Promise<void> {
|
||||
private async writeFinalStatus(
|
||||
exportDate: string,
|
||||
resourceMeta?: DrugDatasetResourceMeta
|
||||
): Promise<void> {
|
||||
// Lazy import to keep module top level free of Lucid
|
||||
const KVStore = (await import('#models/kv_store')).default
|
||||
await KVStore.setValue('drugReference.lastUpdatedExportDate', exportDate)
|
||||
|
||||
// Install-state write-back: when this ingest was kicked off by a curated-tier
|
||||
// install, record an `installed_resources` row so the tier-status math (which
|
||||
// is row-driven for ZIM/map) recognizes the dataset uniformly — the home-tile
|
||||
// gate and the tier "installed" badge both read these rows. resource_type
|
||||
// 'dataset' was widened onto the model in the foundational slice. The row's
|
||||
// `version` is the openFDA export_date (the real freshness key), not the
|
||||
// manifest placeholder. Manual downloads pass no resourceMeta → no row, which
|
||||
// is correct: install-state belongs to the curated-tier path only.
|
||||
if (!resourceMeta) return
|
||||
try {
|
||||
const { default: InstalledResource } = await import('#models/installed_resource')
|
||||
const { DateTime } = await import('luxon')
|
||||
const totalBytes = await this.totalDownloadedBytes()
|
||||
await InstalledResource.updateOrCreate(
|
||||
{ resource_id: resourceMeta.resourceId, resource_type: 'dataset' },
|
||||
{
|
||||
version: exportDate,
|
||||
collection_ref: resourceMeta.collectionRef,
|
||||
url: 'https://api.fda.gov/download.json',
|
||||
// No single on-disk file — the parts are deleted after ingest; the
|
||||
// installed artifact is the DB table. Record the staging dir for
|
||||
// provenance; uninstall keys off resource_id, never this path.
|
||||
file_path: STORAGE_BASE,
|
||||
file_size_bytes: totalBytes,
|
||||
installed_at: DateTime.now(),
|
||||
}
|
||||
)
|
||||
logger.info(
|
||||
`[IngestDrugDataJob] Wrote installed_resources row for ${resourceMeta.resourceId} (export_date=${exportDate})`
|
||||
)
|
||||
} catch (err) {
|
||||
// A failed row write must NOT abort a completed ingest — the data is
|
||||
// already searchable. Log loud; the tier badge will simply read
|
||||
// not-installed until the next install reconcile.
|
||||
logger.error(
|
||||
`[IngestDrugDataJob] Failed to write installed_resources row for ${resourceMeta.resourceId}: ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Sum the recorded part sizes from the download-state KV marker, for the
|
||||
* `installed_resources.file_size_bytes` column. Best-effort: the parts are
|
||||
* deleted right after a full ingest, so the marker (written before deletion) is
|
||||
* the only durable size source. Returns null when the marker is absent/empty so
|
||||
* the column stays NULL rather than reporting a wrong 0.
|
||||
*/
|
||||
private async totalDownloadedBytes(): Promise<number | null> {
|
||||
try {
|
||||
const KVStore = (await import('#models/kv_store')).default
|
||||
const { parseDownloadState } = await import('../../util/drug_labels.js')
|
||||
const marker = parseDownloadState(await KVStore.getValue('drugReference.downloadState'))
|
||||
if (!marker || marker.parts.length === 0) return null
|
||||
const sum = marker.parts.reduce((acc, p) => acc + (p.bytes || 0), 0)
|
||||
return sum > 0 ? sum : null
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -98,15 +98,28 @@ export class CollectionManifestService {
|
|||
const spec = await this.getSpecWithFallback<ZimCategoriesSpec>('zim_categories')
|
||||
if (!spec) return []
|
||||
|
||||
const installedResources = await InstalledResource.query().where('resource_type', 'zim')
|
||||
// Include 'dataset' rows alongside 'zim' so a curated tier carrying the FDA
|
||||
// drug dataset reads "installed" once the ingest writes its row (the tier-
|
||||
// status math below treats every resolved resource id uniformly — a dataset
|
||||
// resource is just another id to account for).
|
||||
const installedResources = await InstalledResource.query().whereIn('resource_type', [
|
||||
'zim',
|
||||
'dataset',
|
||||
])
|
||||
const installedMap = new Map(installedResources.map((r) => [r.resource_id, r]))
|
||||
|
||||
// In-flight ZIM download resource IDs from the BullMQ queue. Used to
|
||||
// surface the user's tier intent immediately on submit, before any single
|
||||
// file has finished downloading. Failed jobs are excluded so a stuck
|
||||
// queue entry doesn't keep claiming the user's pick forever.
|
||||
// In-flight ZIM + dataset download resource IDs from the BullMQ queues. Used
|
||||
// to surface the user's tier intent immediately on submit, before any single
|
||||
// file has finished downloading. Failed jobs are excluded so a stuck queue
|
||||
// entry doesn't keep claiming the user's pick forever.
|
||||
const inFlightIds = await this.getInFlightZimResourceIds()
|
||||
|
||||
// Whether the in-flight drug dataset is in its INGEST (indexing) phase — the
|
||||
// download finished and the heavy ingest is running. Lets the wizard card
|
||||
// flip "(downloading)" → "(indexing)" at the handoff (Req 7) instead of
|
||||
// showing a stale "downloading" through the long index.
|
||||
const drugIndexing = await this.isDrugDatasetIndexing()
|
||||
|
||||
return spec.categories.map((category) => {
|
||||
const installedTierSlug = this.getInstalledTierForCategory(category.tiers, installedMap)
|
||||
const downloadingTierSlug = this.getDownloadingTierForCategory(
|
||||
|
|
@ -115,10 +128,47 @@ export class CollectionManifestService {
|
|||
inFlightIds,
|
||||
installedTierSlug
|
||||
)
|
||||
return { ...category, installedTierSlug, downloadingTierSlug }
|
||||
// Only mark "indexing" when this category's downloading tier actually
|
||||
// carries a dataset resource that is the one indexing.
|
||||
const downloadingTierIndexing =
|
||||
drugIndexing && downloadingTierSlug
|
||||
? CollectionManifestService.resolveTierResources(
|
||||
category.tiers.find((t) => t.slug === downloadingTierSlug)!,
|
||||
category.tiers
|
||||
).some((r) => r.type === 'dataset')
|
||||
: false
|
||||
return { ...category, installedTierSlug, downloadingTierSlug, downloadingTierIndexing }
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* True when the FDA drug dataset's INGEST is in flight while its DOWNLOAD is
|
||||
* not — i.e. the handoff into the indexing phase. Drives the wizard card's
|
||||
* "(indexing)" label. Defensive: any queue read failure returns false (the card
|
||||
* just keeps showing "(downloading)") rather than breaking the categories list.
|
||||
*/
|
||||
private async isDrugDatasetIndexing(): Promise<boolean> {
|
||||
try {
|
||||
const { DownloadDrugDataJob } = await import('#jobs/download_drug_data_job')
|
||||
const { IngestDrugDataJob } = await import('#jobs/ingest_drug_data_job')
|
||||
const queueService = QueueService.getInstance()
|
||||
|
||||
const ingestQueue = queueService.getQueue(IngestDrugDataJob.queue)
|
||||
const ingestJobs = await ingestQueue.getJobs(['active', 'waiting', 'delayed'])
|
||||
if (ingestJobs.length === 0) return false
|
||||
|
||||
const downloadQueue = queueService.getQueue(DownloadDrugDataJob.queue)
|
||||
const downloadJobs = await downloadQueue.getJobs(['active', 'waiting', 'delayed'])
|
||||
return downloadJobs.length === 0
|
||||
} catch (error: any) {
|
||||
logger.warn(
|
||||
'[CollectionManifestService] Could not determine drug indexing state:',
|
||||
error?.message || error
|
||||
)
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
private async getInFlightZimResourceIds(): Promise<Set<string>> {
|
||||
const ids = new Set<string>()
|
||||
try {
|
||||
|
|
@ -134,9 +184,47 @@ export class CollectionManifestService {
|
|||
// unreachable — just report no in-flight downloads.
|
||||
logger.warn('[CollectionManifestService] Could not read download queue:', error?.message || error)
|
||||
}
|
||||
|
||||
// Also surface an in-flight curated-tier drug-dataset install. The drug
|
||||
// download/ingest run on their own queues (not RunDownloadJob), carrying the
|
||||
// dataset's resource id in resourceMeta — read it so the wizard shows the
|
||||
// Medicine tier as "downloading" the moment the user opts in, mirroring the
|
||||
// ZIM behaviour. Scanned independently so a missing drug queue can't blank
|
||||
// the ZIM in-flight set above.
|
||||
await this.addInFlightDrugDatasetId(ids)
|
||||
|
||||
return ids
|
||||
}
|
||||
|
||||
/**
|
||||
* Add the FDA drug dataset's manifest resource id to `ids` when its download or
|
||||
* ingest is in flight. The dataset only counts as "downloading" while no
|
||||
* install-state row exists yet — once ingest writes the row it is "installed"
|
||||
* (handled by the installedMap), so a still-running ingest correctly reads as
|
||||
* the in-flight tier intent here. resourceMeta is only present on a curated-tier
|
||||
* install, so a manual (non-tier) drug download never claims a tier slug.
|
||||
*/
|
||||
private async addInFlightDrugDatasetId(ids: Set<string>): Promise<void> {
|
||||
try {
|
||||
const { DownloadDrugDataJob } = await import('#jobs/download_drug_data_job')
|
||||
const { IngestDrugDataJob } = await import('#jobs/ingest_drug_data_job')
|
||||
const queueService = QueueService.getInstance()
|
||||
for (const queueName of [DownloadDrugDataJob.queue, IngestDrugDataJob.queue]) {
|
||||
const queue = queueService.getQueue(queueName)
|
||||
const jobs = await queue.getJobs(['waiting', 'active', 'delayed'])
|
||||
for (const job of jobs) {
|
||||
const resourceId = job.data?.resourceMeta?.resourceId
|
||||
if (typeof resourceId === 'string') ids.add(resourceId)
|
||||
}
|
||||
}
|
||||
} catch (error: any) {
|
||||
logger.warn(
|
||||
'[CollectionManifestService] Could not read drug dataset queue:',
|
||||
error?.message || error
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Highest tier whose every resource is installed OR has an in-flight
|
||||
* download. Returns undefined when there are no in-flight downloads for this
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import { RunDownloadJob } from '#jobs/run_download_job'
|
|||
import { RunExtractPmtilesJob } from '#jobs/run_extract_pmtiles_job'
|
||||
import type { RunExtractPmtilesJobParams } from '#jobs/run_extract_pmtiles_job'
|
||||
import { DownloadModelJob } from '#jobs/download_model_job'
|
||||
import { DownloadDrugDataJob } from '#jobs/download_drug_data_job'
|
||||
import { DownloadJobWithProgress, DownloadProgressData } from '../../types/downloads.js'
|
||||
import type { Job, Queue } from 'bullmq'
|
||||
import { normalize } from 'path'
|
||||
|
|
@ -51,10 +52,11 @@ export class DownloadService {
|
|||
|
||||
async listDownloadJobs(filetype?: string): Promise<DownloadJobWithProgress[]> {
|
||||
const modelQueue = this.queueService.getQueue(DownloadModelJob.queue)
|
||||
const [fileTagged, extractTagged, modelJobs] = await Promise.all([
|
||||
const [fileTagged, extractTagged, modelJobs, drugTagged] = await Promise.all([
|
||||
this.fetchJobsWithStates(RunDownloadJob.queue),
|
||||
this.fetchJobsWithStates(RunExtractPmtilesJob.queue),
|
||||
modelQueue.getJobs(['waiting', 'active', 'delayed', 'failed']),
|
||||
this.fetchJobsWithStates(DownloadDrugDataJob.queue),
|
||||
])
|
||||
|
||||
const fileDownloads = fileTagged.map(({ job, state }) => {
|
||||
|
|
@ -101,7 +103,32 @@ export class DownloadService {
|
|||
failedReason: job.failedReason || undefined,
|
||||
}))
|
||||
|
||||
const allDownloads = [...fileDownloads, ...extractDownloads, ...modelDownloads]
|
||||
// FDA drug dataset — DOWNLOAD phase only. The job fans the manifest's N
|
||||
// partitions into continuations under auto-generated jobIds, but the user
|
||||
// sees ONE download, so collapse to a single card by keeping ONLY the
|
||||
// deterministic jobId job (the part currently downloading lives under it; the
|
||||
// continuations are an internal implementation detail). The heavy ingest is
|
||||
// deliberately EXCLUDED here — it stays on the IngestStatus surface.
|
||||
const drugDownloads = drugTagged
|
||||
.filter(({ job }) => job.id?.toString() === DownloadDrugDataJob.jobId)
|
||||
.map(({ job, state }) => {
|
||||
const parsed = this.parseProgress(job.progress)
|
||||
return {
|
||||
jobId: job.id!.toString(),
|
||||
url: 'https://api.fda.gov/download.json',
|
||||
progress: parsed.percent,
|
||||
filepath: job.data.currentPartName || 'FDA Drug Reference',
|
||||
filetype: 'drug-data',
|
||||
title: 'FDA Drug Reference',
|
||||
downloadedBytes: parsed.downloadedBytes,
|
||||
totalBytes: parsed.totalBytes,
|
||||
lastProgressTime: parsed.lastProgressTime,
|
||||
status: state,
|
||||
failedReason: job.failedReason || undefined,
|
||||
}
|
||||
})
|
||||
|
||||
const allDownloads = [...fileDownloads, ...extractDownloads, ...modelDownloads, ...drugDownloads]
|
||||
const filtered = allDownloads.filter((job) => !filetype || job.filetype === filetype)
|
||||
|
||||
return filtered.sort((a, b) => {
|
||||
|
|
@ -116,6 +143,7 @@ export class DownloadService {
|
|||
RunDownloadJob.queue,
|
||||
RunExtractPmtilesJob.queue,
|
||||
DownloadModelJob.queue,
|
||||
DownloadDrugDataJob.queue,
|
||||
]) {
|
||||
const queue = this.queueService.getQueue(queueName)
|
||||
const job = await queue.getJob(jobId)
|
||||
|
|
@ -159,9 +187,42 @@ export class DownloadService {
|
|||
return await this._cancelModelDownloadJob(jobId, modelJob, modelQueue)
|
||||
}
|
||||
|
||||
// FDA drug dataset: cancel is matched on the deterministic jobId only (the
|
||||
// single card the aggregator shows). The continuation parts run under
|
||||
// auto-generated ids, so cancelling must stop the whole CHAIN, not just the
|
||||
// current part — otherwise the next continuation fires after we remove one.
|
||||
if (jobId === DownloadDrugDataJob.jobId) {
|
||||
return await this._cancelDrugDownloadJob()
|
||||
}
|
||||
|
||||
return { success: true, message: 'Job not found (may have already completed)' }
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancel the FDA drug-data download.
|
||||
*
|
||||
* The drug job has no Redis cancel-signal / AbortController (unlike
|
||||
* RunDownloadJob), and it self-continues into the next part under a fresh jobId.
|
||||
* So a v1 cancel that only removed the current job would leave the next
|
||||
* continuation to fire. Instead we obliterate the single-purpose drug-download
|
||||
* queue (force = removes the active/locked job too), which drops the current
|
||||
* part AND every queued continuation in one shot. Scoped to the drug-download
|
||||
* queue only — the ingest queue and everything else are untouched. The on-disk
|
||||
* parts are intentionally LEFT in place: they're resumable, and a re-trigger
|
||||
* picks up from the .tmp rather than re-downloading. (Tracked as the v1 choice
|
||||
* for cancel depth — full cross-process signal cancel is a follow-up.)
|
||||
*/
|
||||
private async _cancelDrugDownloadJob(): Promise<{ success: boolean; message: string }> {
|
||||
const queue = this.queueService.getQueue(DownloadDrugDataJob.queue)
|
||||
try {
|
||||
await queue.obliterate({ force: true })
|
||||
return { success: true, message: 'Drug data download cancelled' }
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err)
|
||||
return { success: false, message: `Could not cancel drug data download: ${msg}` }
|
||||
}
|
||||
}
|
||||
|
||||
private async _cancelExtractJob(
|
||||
jobId: string,
|
||||
job: Job<RunExtractPmtilesJobParams>,
|
||||
|
|
|
|||
|
|
@ -1,9 +1,18 @@
|
|||
import db from '@adonisjs/lucid/services/db'
|
||||
import logger from '@adonisjs/core/services/logger'
|
||||
import { rm } from 'node:fs/promises'
|
||||
import type { Job } from 'bullmq'
|
||||
import { QueueService } from './queue_service.js'
|
||||
import { DownloadDrugDataJob } from '#jobs/download_drug_data_job'
|
||||
import { DownloadDrugDataJob, STORAGE_BASE } from '#jobs/download_drug_data_job'
|
||||
import { IngestDrugDataJob } from '#jobs/ingest_drug_data_job'
|
||||
|
||||
/**
|
||||
* Manifest resource id for the FDA drug dataset — the `installed_resources`
|
||||
* row's `resource_id`, matching the `medicine-standard` tier manifest entry.
|
||||
* Single source of truth so the install write-back, the tier-status math, and
|
||||
* uninstall all agree on the same id.
|
||||
*/
|
||||
export const DRUG_DATASET_RESOURCE_ID = 'openfda-drug-labels'
|
||||
import {
|
||||
normalizeDrugName,
|
||||
parseDownloadState,
|
||||
|
|
@ -11,6 +20,7 @@ import {
|
|||
resolveExpectedTotal,
|
||||
resolveIngestRecordsShown,
|
||||
summarizeJobError,
|
||||
isExportDateNewer,
|
||||
} from '../../util/drug_labels.js'
|
||||
import KVStore from '#models/kv_store'
|
||||
import { parseCompareIds, MAX_COMPARE } from '../../util/compare_ids.js'
|
||||
|
|
@ -397,6 +407,32 @@ export class DrugReferenceService {
|
|||
return DownloadDrugDataJob.dispatch(true)
|
||||
}
|
||||
|
||||
/**
|
||||
* Freshness check for the daily 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).
|
||||
*/
|
||||
async checkForUpdate(): Promise<{
|
||||
updateAvailable: boolean
|
||||
latestExportDate: string | null
|
||||
currentExportDate: string | null
|
||||
}> {
|
||||
const current = await KVStore.getValue('drugReference.lastUpdatedExportDate')
|
||||
const manifest = await DownloadDrugDataJob.fetchManifest()
|
||||
const latest = manifest.export_date ?? null
|
||||
return {
|
||||
updateAvailable: isExportDateNewer(latest ?? '', current),
|
||||
latestExportDate: latest,
|
||||
currentExportDate: current,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
|
|
@ -457,6 +493,108 @@ export class DrugReferenceService {
|
|||
return { ...result, nothingDownloaded: false }
|
||||
}
|
||||
|
||||
/**
|
||||
* Uninstall the offline FDA drug dataset — the curated-tier "remove" path.
|
||||
*
|
||||
* Mirrors how ZimService.delete() reverses a ZIM install: stop anything that
|
||||
* could re-create the data, delete the on-disk artifacts, drop the data, and
|
||||
* remove the install-state row so the tier reads not-installed and the home
|
||||
* tiles auto-hide (the same cascade as a ZIM delete).
|
||||
*
|
||||
* Order matters:
|
||||
* 1. Obliterate BOTH drug queues (force = removes active/locked jobs too) so a
|
||||
* running download/ingest can't write rows back in after we clear them.
|
||||
* Scoped to the two single-purpose drug queues — nothing else is touched.
|
||||
* 2. Delete the on-disk parts dir (STORAGE_BASE is a FIXED constant, never a
|
||||
* client value — no path-traversal surface). Usually empty after a full
|
||||
* ingest; non-empty only when uninstalling a downloaded-not-yet-ingested
|
||||
* state.
|
||||
* 3. TRUNCATE drug_labels (not DROP) — preserves the schema + FULLTEXT
|
||||
* indexes so a later reinstall re-ingests without a migration.
|
||||
* 4. Clear the KV markers (download-state + last-updated export_date).
|
||||
* 5. Delete the `installed_resources` 'dataset' row.
|
||||
*
|
||||
* Every step is best-effort and logged: a partial failure still removes as much
|
||||
* as it can and reports what it did, rather than leaving a half-uninstalled
|
||||
* state with no signal.
|
||||
*/
|
||||
async uninstall(): Promise<{
|
||||
success: boolean
|
||||
rowsDropped: number
|
||||
message: string
|
||||
}> {
|
||||
const rowsBefore = await this.rowCount()
|
||||
const errors: string[] = []
|
||||
|
||||
// 1. Stop in-flight jobs on both drug queues (force removes locked/active).
|
||||
for (const queueName of [DownloadDrugDataJob.queue, IngestDrugDataJob.queue]) {
|
||||
try {
|
||||
await QueueService.getInstance().getQueue(queueName).obliterate({ force: true })
|
||||
logger.info(`[DrugReferenceService] obliterated ${queueName} for uninstall`)
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err)
|
||||
logger.warn(`[DrugReferenceService] obliterate ${queueName} failed: ${msg}`)
|
||||
errors.push(`queue ${queueName}: ${msg}`)
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Delete the on-disk parts directory. STORAGE_BASE is a fixed module const
|
||||
// (never a client filename), so this is not a path-traversal surface.
|
||||
try {
|
||||
await rm(STORAGE_BASE, { recursive: true, force: true })
|
||||
logger.info(`[DrugReferenceService] removed on-disk parts dir ${STORAGE_BASE}`)
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err)
|
||||
logger.warn(`[DrugReferenceService] could not remove ${STORAGE_BASE}: ${msg}`)
|
||||
errors.push(`storage: ${msg}`)
|
||||
}
|
||||
|
||||
// 3. Drop the searchable data. TRUNCATE keeps schema + the FULLTEXT indexes
|
||||
// so reinstall re-ingests cleanly with no migration.
|
||||
try {
|
||||
await db.rawQuery('TRUNCATE TABLE drug_labels')
|
||||
logger.info('[DrugReferenceService] truncated drug_labels')
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err)
|
||||
logger.error(`[DrugReferenceService] TRUNCATE drug_labels failed: ${msg}`)
|
||||
errors.push(`truncate: ${msg}`)
|
||||
}
|
||||
|
||||
// 4. Clear KV markers.
|
||||
try {
|
||||
await KVStore.clearValue('drugReference.downloadState')
|
||||
await KVStore.clearValue('drugReference.lastUpdatedExportDate')
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err)
|
||||
logger.warn(`[DrugReferenceService] clearing KV markers failed: ${msg}`)
|
||||
errors.push(`kv: ${msg}`)
|
||||
}
|
||||
|
||||
// 5. Remove the install-state row so the tier reads not-installed.
|
||||
try {
|
||||
const { default: InstalledResource } = await import('#models/installed_resource')
|
||||
await InstalledResource.query()
|
||||
.where('resource_type', 'dataset')
|
||||
.where('resource_id', DRUG_DATASET_RESOURCE_ID)
|
||||
.delete()
|
||||
logger.info('[DrugReferenceService] deleted installed_resources dataset row')
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err)
|
||||
logger.warn(`[DrugReferenceService] deleting install row failed: ${msg}`)
|
||||
errors.push(`install-row: ${msg}`)
|
||||
}
|
||||
|
||||
const rowsDropped = rowsBefore - (await this.rowCount())
|
||||
const success = errors.length === 0
|
||||
return {
|
||||
success,
|
||||
rowsDropped,
|
||||
message: success
|
||||
? `Uninstalled FDA drug reference (${rowsDropped} labels removed).`
|
||||
: `Uninstall completed with ${errors.length} issue(s): ${errors.join('; ')}`,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the canonical deterministic job for a phase's queue, falling back to
|
||||
* the most-progressed auto-id continuation when the deterministic job is
|
||||
|
|
|
|||
|
|
@ -255,8 +255,13 @@ export class ZimService {
|
|||
|
||||
const allResources = CollectionManifestService.resolveTierResources(tier, category.tiers)
|
||||
|
||||
// Filter out already installed
|
||||
const installed = await InstalledResource.query().where('resource_type', 'zim')
|
||||
// Filter out already installed. Includes 'dataset' rows (the FDA drug labels)
|
||||
// alongside 'zim' so an installed dataset is filtered out here — without it,
|
||||
// a re-select of an installed Medicine tier would re-dispatch the ~1.7 GB drug
|
||||
// download every time (the dataset branch's own getIngestStatus guard is a
|
||||
// second line of defence, but this keeps the filter symmetric with the row-
|
||||
// driven tier-status math).
|
||||
const installed = await InstalledResource.query().whereIn('resource_type', ['zim', 'dataset'])
|
||||
const installedIds = new Set(installed.map((r) => r.resource_id))
|
||||
const toDownload = allResources.filter((r) => !installedIds.has(r.id))
|
||||
|
||||
|
|
@ -267,13 +272,10 @@ export class ZimService {
|
|||
for (const resource of toDownload) {
|
||||
// A `dataset` resource (e.g. the FDA drug labels) is DB-ingested, not a
|
||||
// ZIM file — route it to the drug download+ingest pipeline instead of
|
||||
// RunDownloadJob. The installed-filter above queries resource_type='zim',
|
||||
// so a dataset is never seen as installed and would re-dispatch the ~1.7 GB
|
||||
// download on every tier select. Guard against that with the existing
|
||||
// drug-status signal: skip when the data is already present.
|
||||
// (Planned follow-up: write an InstalledResource 'dataset' row on ingest
|
||||
// 'ready' so the tier-status/badge math recognizes it uniformly — a
|
||||
// maintainer decision still pending, out of scope for this slice.)
|
||||
// RunDownloadJob. The install-state row written on ingest 'ready' (below,
|
||||
// via resourceMeta) is what the installed-filter above and the tier-status
|
||||
// math key off; until that row exists the getIngestStatus guard prevents
|
||||
// re-dispatching the ~1.7 GB download on every tier select.
|
||||
if (resource.type === 'dataset') {
|
||||
const drugReferenceService = new DrugReferenceService()
|
||||
const status = await drugReferenceService.getIngestStatus()
|
||||
|
|
@ -283,8 +285,14 @@ export class ZimService {
|
|||
}
|
||||
// DownloadDrugDataJob.dispatch() is idempotent on its deterministic
|
||||
// jobId — a concurrent in-flight download returns "already running"
|
||||
// without re-adding, so this is safe to call repeatedly.
|
||||
await DownloadDrugDataJob.dispatch()
|
||||
// without re-adding, so this is safe to call repeatedly. The resourceMeta
|
||||
// is threaded through download → ingest so the final ingest pass writes
|
||||
// the `installed_resources` 'dataset' row, making the tier read installed.
|
||||
await DownloadDrugDataJob.dispatch(true, {
|
||||
resourceId: resource.id,
|
||||
version: resource.version,
|
||||
collectionRef: categorySlug,
|
||||
})
|
||||
logger.info('[ZimService] Dispatched drug data download for dataset resource.')
|
||||
continue
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ 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'
|
||||
|
|
@ -150,6 +151,7 @@ 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
|
||||
|
|
@ -183,6 +185,7 @@ 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())
|
||||
|
||||
|
|
@ -196,6 +199,7 @@ 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)
|
||||
|
||||
|
|
|
|||
|
|
@ -75,7 +75,8 @@ const CategoryCard: React.FC<CategoryCardProps> = ({ category, selectedTier, onC
|
|||
)}
|
||||
<span className="text-lime-400 text-sm ml-1">
|
||||
{badgeTier.name}
|
||||
{badgeStatus === 'downloading' && ' (downloading)'}
|
||||
{badgeStatus === 'downloading' &&
|
||||
(category.downloadingTierIndexing ? ' (indexing)' : ' (downloading)')}
|
||||
</span>
|
||||
</div>
|
||||
) : (
|
||||
|
|
|
|||
|
|
@ -115,6 +115,10 @@ export default function Home(props: {
|
|||
system: {
|
||||
services: ServiceSlim[]
|
||||
}
|
||||
// Server-computed: true when the offline FDA drug dataset is installed or
|
||||
// installing (curated Medicine tier). Gates the two medical-reference tiles
|
||||
// below so they only appear once the data exists.
|
||||
drugReferenceInstalled: boolean
|
||||
}) {
|
||||
const items: DashboardItem[] = []
|
||||
const updateInfo = useUpdateAvailable();
|
||||
|
|
@ -155,9 +159,13 @@ export default function Home(props: {
|
|||
// Add Maps as a Core Capability
|
||||
items.push(MAPS_ITEM)
|
||||
|
||||
// Add the offline medical-reference Core Capabilities
|
||||
items.push(DRUG_REFERENCE_ITEM)
|
||||
items.push(CONDITIONS_ITEM)
|
||||
// Add the offline medical-reference tiles only once the FDA drug dataset is
|
||||
// installed (or installing) via the curated Medicine tier. Both tiles read the
|
||||
// same drug_labels table, so they gate together off one server-computed flag.
|
||||
if (props.drugReferenceInstalled) {
|
||||
items.push(DRUG_REFERENCE_ITEM)
|
||||
items.push(CONDITIONS_ITEM)
|
||||
}
|
||||
|
||||
// Add system items
|
||||
items.push(...SYSTEM_ITEMS)
|
||||
|
|
|
|||
|
|
@ -261,6 +261,7 @@ router
|
|||
router.post('/download', [DrugReferenceController, 'download'])
|
||||
router.post('/ingest', [DrugReferenceController, 'ingest'])
|
||||
router.post('/reset-ingest', [DrugReferenceController, 'resetIngest'])
|
||||
router.post('/uninstall', [DrugReferenceController, 'uninstall'])
|
||||
router.get('/ingest-log', [DrugReferenceController, 'ingestLog'])
|
||||
})
|
||||
.prefix('/api/drug-reference')
|
||||
|
|
|
|||
|
|
@ -0,0 +1,81 @@
|
|||
/**
|
||||
* Standalone gate test for the tier-rework pure helpers added in the
|
||||
* drug-reference opt-in-tier rework (PR #1040 follow-up):
|
||||
* - manifestBytesTotal (Active Downloads card totalBytes)
|
||||
* - isExportDateNewer (drug auto-update freshness compare)
|
||||
*
|
||||
* Japa can't boot locally without MySQL/Redis, so these exercise the pure
|
||||
* helpers directly under `node --experimental-strip-types`. Run:
|
||||
* node --experimental-strip-types tests/standalone/drug_tier_rework.standalone.ts
|
||||
*/
|
||||
import assert from 'node:assert/strict'
|
||||
import { manifestBytesTotal, isExportDateNewer } from '../../util/drug_labels.ts'
|
||||
import type { DrugLabelManifest } from '../../types/drug_reference.ts'
|
||||
|
||||
let passed = 0
|
||||
function check(name: string, fn: () => void) {
|
||||
fn()
|
||||
passed++
|
||||
console.log(` ok - ${name}`)
|
||||
}
|
||||
|
||||
const MB = 1024 * 1024
|
||||
|
||||
function manifest(sizes: string[]): DrugLabelManifest {
|
||||
return {
|
||||
export_date: '2025-01-01',
|
||||
total_records: 1000,
|
||||
partitions: sizes.map((s, i) => ({
|
||||
display_name: `part ${i + 1}`,
|
||||
file: `https://download.open.fda.gov/drug/label/p-${i}.json.zip`,
|
||||
size_mb: s,
|
||||
records: 100,
|
||||
})),
|
||||
}
|
||||
}
|
||||
|
||||
// ── manifestBytesTotal ────────────────────────────────────────────────────────
|
||||
check('manifestBytesTotal sums partition MB into bytes', () =>
|
||||
assert.equal(manifestBytesTotal(manifest(['100', '200', '50'])), 350 * MB))
|
||||
check('manifestBytesTotal treats a non-numeric size as 0 (no NaN)', () =>
|
||||
assert.equal(manifestBytesTotal(manifest(['100', 'oops', '50'])), 150 * MB))
|
||||
check('manifestBytesTotal treats a negative/zero size as 0', () =>
|
||||
assert.equal(manifestBytesTotal(manifest(['100', '0', '-5'])), 100 * MB))
|
||||
check('manifestBytesTotal of an empty partition list is 0', () =>
|
||||
assert.equal(manifestBytesTotal(manifest([])), 0))
|
||||
|
||||
// ── isExportDateNewer ─────────────────────────────────────────────────────────
|
||||
// No baseline → any valid candidate is "newer" (first install with no marker).
|
||||
check('isExportDateNewer: empty baseline is always newer', () =>
|
||||
assert.equal(isExportDateNewer('2025-06-01', null), true))
|
||||
check('isExportDateNewer: empty-string baseline is always newer', () =>
|
||||
assert.equal(isExportDateNewer('2025-06-01', ''), true))
|
||||
// Empty candidate → nothing to update to.
|
||||
check('isExportDateNewer: empty candidate is never newer', () =>
|
||||
assert.equal(isExportDateNewer('', '2025-06-01'), false))
|
||||
// Identical → already current.
|
||||
check('isExportDateNewer: identical dates are not newer', () =>
|
||||
assert.equal(isExportDateNewer('2025-06-01', '2025-06-01'), false))
|
||||
// ISO YYYY-MM-DD — both Date.parse paths and lexicographic agree.
|
||||
check('isExportDateNewer: ISO later date is newer', () =>
|
||||
assert.equal(isExportDateNewer('2025-06-02', '2025-06-01'), true))
|
||||
check('isExportDateNewer: ISO earlier date is not newer', () =>
|
||||
assert.equal(isExportDateNewer('2025-05-31', '2025-06-01'), false))
|
||||
// Cross-month / cross-year ISO ordering.
|
||||
check('isExportDateNewer: ISO across year boundary', () =>
|
||||
assert.equal(isExportDateNewer('2026-01-01', '2025-12-31'), true))
|
||||
// US M/D/YYYY format — Date.parse handles this; a naive lexicographic compare
|
||||
// would get it WRONG ('1/2/2026' < '12/31/2025' lexically). This proves the
|
||||
// Date.parse-first path is doing the work.
|
||||
check('isExportDateNewer: US-format newer date (Date.parse path)', () =>
|
||||
assert.equal(isExportDateNewer('1/2/2026', '12/31/2025'), true))
|
||||
check('isExportDateNewer: US-format older date (Date.parse path)', () =>
|
||||
assert.equal(isExportDateNewer('12/30/2025', '12/31/2025'), false))
|
||||
// Zero-padded compact YYYYMMDD — Date.parse rejects it, lexicographic fallback
|
||||
// is correct for this zero-padded ordered form.
|
||||
check('isExportDateNewer: compact YYYYMMDD newer (lexicographic fallback)', () =>
|
||||
assert.equal(isExportDateNewer('20260102', '20251231'), true))
|
||||
check('isExportDateNewer: compact YYYYMMDD older (lexicographic fallback)', () =>
|
||||
assert.equal(isExportDateNewer('20251230', '20251231'), false))
|
||||
|
||||
console.log(`\nAll ${passed} standalone assertions passed.`)
|
||||
|
|
@ -75,6 +75,11 @@ export type CategoryWithStatus = SpecCategory & {
|
|||
// picked something larger and downloads are still running. Lets the UI show
|
||||
// the user's actual intent during the (often long) download window.
|
||||
downloadingTierSlug?: string
|
||||
// True when the in-flight resource for `downloadingTierSlug` is a DB-ingested
|
||||
// dataset (the FDA drug labels) whose download has finished and the heavy
|
||||
// ingest is running — so the card can flip "(downloading)" → "(indexing)" at
|
||||
// the handoff instead of looking stuck. Only meaningful with downloadingTierSlug.
|
||||
downloadingTierIndexing?: boolean
|
||||
}
|
||||
|
||||
export type CollectionWithStatus = SpecCollection & {
|
||||
|
|
|
|||
|
|
@ -65,6 +65,27 @@ export interface DownloadStateMarker {
|
|||
completedAtMs: number
|
||||
}
|
||||
|
||||
// ─── Install-state metadata (tier-installer → InstalledResource row) ──────────
|
||||
|
||||
/**
|
||||
* Identity an install writes to the `installed_resources` table when the drug
|
||||
* dataset reaches `ready`, so the curated-tier "installed" math (which is row-
|
||||
* driven for ZIM/map) recognizes the dataset uniformly. Threaded from the tier
|
||||
* installer (the manifest `dataset` resource) through the download job into the
|
||||
* ingest job, which writes the row. Absent on a manual download (the standalone
|
||||
* "Download FDA data" path), in which case no row is written — the install-state
|
||||
* row only exists when the install came through a curated tier.
|
||||
*
|
||||
* `version` is filled with the dataset's openFDA `export_date` at write time
|
||||
* (the real freshness key), NOT the manifest's placeholder `version` string.
|
||||
*/
|
||||
export interface DrugDatasetResourceMeta {
|
||||
resourceId: string
|
||||
/** Manifest `version` placeholder; the row's real version is the export_date. */
|
||||
version: string
|
||||
collectionRef: string | null
|
||||
}
|
||||
|
||||
// ─── Job params ───────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
|
|
@ -84,6 +105,12 @@ export interface DownloadDrugDataJobParams {
|
|||
bytesDownloaded?: number
|
||||
currentPartName?: string | null
|
||||
phase?: 'manifest' | 'downloading' | 'downloaded' | 'failed'
|
||||
/**
|
||||
* Install-state identity, present only when the install came through a curated
|
||||
* tier. Carried through the continuation chain and handed to the ingest job so
|
||||
* it can write the `installed_resources` row on `ready`.
|
||||
*/
|
||||
resourceMeta?: DrugDatasetResourceMeta
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -108,6 +135,12 @@ export interface IngestDrugDataJobParams {
|
|||
startRowCount?: number
|
||||
currentPartName?: string | null
|
||||
phase?: 'ingesting' | 'ready' | 'failed'
|
||||
/**
|
||||
* Install-state identity, forwarded from the download job when the install
|
||||
* came through a curated tier. The ingest job writes the `installed_resources`
|
||||
* row on `ready` using this. Absent on a manual ingest.
|
||||
*/
|
||||
resourceMeta?: DrugDatasetResourceMeta
|
||||
}
|
||||
|
||||
// ─── Search result DTO (collapsed by brand+generic) ──────────────────────────
|
||||
|
|
|
|||
|
|
@ -323,6 +323,52 @@ export function partZipPath(storageBase: string, partition: DrugLabelPartition):
|
|||
return path.join(storageBase, path.basename(partition.file))
|
||||
}
|
||||
|
||||
/**
|
||||
* Sum the manifest partitions' `size_mb` into a single byte total for the Active
|
||||
* Downloads card's `totalBytes`. Each partition's `size_mb` is a string of MB in
|
||||
* the openFDA manifest; a non-numeric / missing value contributes 0 so a single
|
||||
* bad partition can't NaN the whole total. Pure (no I/O) so the producer and the
|
||||
* standalone test share one definition.
|
||||
*/
|
||||
export function manifestBytesTotal(manifest: DrugLabelManifest): number {
|
||||
return manifest.partitions.reduce((acc, p) => {
|
||||
const mb = Number(p.size_mb)
|
||||
return acc + (Number.isFinite(mb) && mb > 0 ? mb * 1024 * 1024 : 0)
|
||||
}, 0)
|
||||
}
|
||||
|
||||
/**
|
||||
* Is the manifest's `export_date` newer than the last-ingested one? Drives the
|
||||
* drug auto-update freshness check.
|
||||
*
|
||||
* TODO(maintainer Q3): CONFIRM the openFDA `export_date` string format before
|
||||
* trusting this. We only have it typed as `string` in the code; the format
|
||||
* (e.g. `YYYY-MM-DD` vs `YYYYMMDD` vs `MM/DD/YYYY`) determines whether a plain
|
||||
* lexicographic `>` sorts chronologically. This implementation is defensive: it
|
||||
* FIRST tries Date.parse on both sides and compares timestamps (correct for any
|
||||
* parseable date string, including ISO and US formats); only if EITHER side
|
||||
* fails to parse does it fall back to a trimmed lexicographic compare (correct
|
||||
* for zero-padded ISO `YYYY-MM-DD` / `YYYYMMDD`). The fallback can misfire on a
|
||||
* non-ISO, non-parseable format — hence the maintainer confirmation. Either way
|
||||
* it never throws and treats an unset/empty `last` as "newer" (a first install
|
||||
* with no baseline should update). Pure: no I/O, fully unit-testable.
|
||||
*/
|
||||
export function isExportDateNewer(latest: string, last: string | null | undefined): boolean {
|
||||
const l = (latest ?? '').trim()
|
||||
const prev = (last ?? '').trim()
|
||||
if (l === '') return false // no candidate → nothing to update to
|
||||
if (prev === '') return true // no baseline → treat any valid candidate as newer
|
||||
if (l === prev) return false // identical → already current
|
||||
|
||||
const lt = Date.parse(l)
|
||||
const pt = Date.parse(prev)
|
||||
if (Number.isFinite(lt) && Number.isFinite(pt)) {
|
||||
return lt > pt
|
||||
}
|
||||
// Fallback: lexicographic. Correct only for zero-padded ISO-ordered strings.
|
||||
return l > prev
|
||||
}
|
||||
|
||||
/**
|
||||
* Defensively parse the `drugReference.downloadState` KV marker.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -61,6 +61,15 @@
|
|||
"description": "NIH's consumer health encyclopedia - diseases, conditions, drugs, supplements",
|
||||
"url": "https://download.kiwix.org/zim/zimit/medlineplus.gov_en_all_2025-01.zim",
|
||||
"size_mb": 1800
|
||||
},
|
||||
{
|
||||
"id": "openfda-drug-labels",
|
||||
"type": "dataset",
|
||||
"version": "2025-01",
|
||||
"title": "FDA Drug Reference",
|
||||
"description": "Offline FDA drug labels - search by drug name",
|
||||
"url": "https://api.fda.gov/download.json",
|
||||
"size_mb": 1700
|
||||
}
|
||||
]
|
||||
},
|
||||
|
|
|
|||
Loading…
Reference in New Issue