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, 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,
  deriveIngestPhase,
  resolveExpectedTotal,
  resolveIngestRecordsShown,
  summarizeJobError,
  isExportDateNewer,
} from '../../util/drug_labels.js'
import KVStore from '#models/kv_store'
import { parseCompareIds, MAX_COMPARE } from '../../util/compare_ids.js'
import type {
  DrugSearchResult,
  DrugLabelDetail,
  DrugIngestStatus,
  DrugPhaseState,
  DrugDownloadStatus,
  DrugIngestPhaseStatus,
  DrugInteractionEntry,
} from '../../types/drug_reference.js'

/**
 * Drug Reference v1 — service layer.
 *
 * Exposes search (collapsed by brand+generic), detail fetch, ingest trigger,
 * and ingest status. Mirrors the shape of DownloadService/ZimService.
 */
export class DrugReferenceService {
  /**
   * Search for drug labels, collapsed by (brand_name, generic_name).
   *
   * Each distinct (brand_name, generic_name) pair returns ONE result — a
   * representative row id (MIN(id)) and a labelCount of how many set_ids
   * collapsed into it. This is the locked UX decision.
   *
   * Scope:
   *   'name'       (default) — existing path: MATCH(searchable_name) on brand+generic.
   *   'indication' — new path: MATCH(searchable_name, indications) on the combined
   *                  ft_drug_labels_name_indications index so users can search by
   *                  what a drug treats ("heartburn", "high blood pressure").
   *
   * Strategy (both scopes):
   *   1. FULLTEXT path: MATCH(cols) AGAINST(? IN NATURAL LANGUAGE MODE)
   *      — relevance-ranked, requires >= 3 chars (innodb_ft_min_token_size = 3).
   *   2. LIKE fallback: query < 3 chars OR FULLTEXT throws → LIKE '%term%'.
   *   3. Both paths apply the optional product_type filter and GROUP BY collapse.
   */
  async search(
    query: string,
    options: {
      productType?: string
      route?: string
      sort?: 'relevance' | 'name'
      limit?: number
      offset?: number
      scope?: 'name' | 'indication'
    }
  ): Promise<DrugSearchResult[]> {
    const limit = options.limit ?? 50
    const offset = options.offset ?? 0
    const scope = options.scope ?? 'name'
    const normalized = normalizeDrugName(query, null) ?? query.trim()

    if (!normalized || normalized.length === 0) return []

    const useLike = normalized.length < 3

    if (scope === 'indication') {
      if (!useLike) {
        try {
          return await this.searchIndicationFulltext(normalized, options.productType, limit, offset)
        } catch (err) {
          logger.warn(
            `[DrugReferenceService] FULLTEXT indication search failed, falling back to LIKE: ${
              err instanceof Error ? err.message : String(err)
            }`
          )
        }
      }
      return await this.searchIndicationLike(normalized, options.productType, limit, offset)
    }

    if (!useLike) {
      // FULLTEXT path
      try {
        return await this.searchFulltext(normalized, options, limit, offset)
      } catch (err) {
        logger.warn(
          `[DrugReferenceService] FULLTEXT search failed, falling back to LIKE: ${
            err instanceof Error ? err.message : String(err)
          }`
        )
      }
    }

    // LIKE fallback
    return await this.searchLike(normalized, options, limit, offset)
  }

  private async searchFulltext(
    normalized: string,
    opts: { productType?: string; route?: string; sort?: 'relevance' | 'name' },
    limit: number,
    offset: number
  ): Promise<DrugSearchResult[]> {
    let sql = `
      SELECT
        MIN(id) AS id,
        brand_name,
        generic_name,
        MIN(manufacturer) AS manufacturer,
        MIN(route) AS route,
        MIN(product_type) AS product_type,
        COUNT(*) AS labelCount,
        MAX(MATCH(searchable_name) AGAINST(? IN NATURAL LANGUAGE MODE)) AS relevance
      FROM drug_labels
      WHERE MATCH(searchable_name) AGAINST(? IN NATURAL LANGUAGE MODE)
    `
    const bindings: unknown[] = [normalized, normalized]

    if (opts.productType) {
      sql += ' AND product_type = ?'
      bindings.push(opts.productType)
    }
    if (opts.route) {
      // `route` is a comma-joined uppercase list (e.g. "ORAL, TOPICAL").
      sql += ' AND route LIKE ?'
      bindings.push(`%${opts.route.toUpperCase()}%`)
    }

    sql += `
      GROUP BY brand_name, generic_name
      ORDER BY ${opts.sort === 'name' ? 'COALESCE(brand_name, generic_name) ASC' : 'relevance DESC'}
      LIMIT ? OFFSET ?
    `
    bindings.push(limit, offset)

    const rows = await db.rawQuery(sql, bindings)
    return this.mapSearchRows(rows[0])
  }

  private async searchLike(
    normalized: string,
    opts: { productType?: string; route?: string; sort?: 'relevance' | 'name' },
    limit: number,
    offset: number
  ): Promise<DrugSearchResult[]> {
    const term = `%${normalized}%`
    let sql = `
      SELECT
        MIN(id) AS id,
        brand_name,
        generic_name,
        MIN(manufacturer) AS manufacturer,
        MIN(route) AS route,
        MIN(product_type) AS product_type,
        COUNT(*) AS labelCount
      FROM drug_labels
      WHERE (searchable_name LIKE ? OR brand_name LIKE ?)
    `
    const bindings: unknown[] = [term, term]

    if (opts.productType) {
      sql += ' AND product_type = ?'
      bindings.push(opts.productType)
    }
    if (opts.route) {
      sql += ' AND route LIKE ?'
      bindings.push(`%${opts.route.toUpperCase()}%`)
    }

    sql += `
      GROUP BY brand_name, generic_name
      ORDER BY brand_name ASC
      LIMIT ? OFFSET ?
    `
    bindings.push(limit, offset)

    const rows = await db.rawQuery(sql, bindings)
    return this.mapSearchRows(rows[0])
  }

  /**
   * FULLTEXT indication-scope search.
   *
   * MATCHes over (searchable_name, indications) — must exactly match the
   * ft_drug_labels_name_indications index column list. The MAX(MATCH …) pattern
   * is LOAD-BEARING: MySQL 8.0 ONLY_FULL_GROUP_BY rejects a bare MATCH() in
   * SELECT when GROUP BY is in effect; wrapping in MAX() makes it an aggregate
   * and satisfies the mode constraint.
   */
  private async searchIndicationFulltext(
    normalized: string,
    productType: string | undefined,
    limit: number,
    offset: number
  ): Promise<DrugSearchResult[]> {
    let sql = `
      SELECT
        MIN(id) AS id,
        brand_name,
        generic_name,
        MIN(manufacturer) AS manufacturer,
        MIN(route) AS route,
        MIN(product_type) AS product_type,
        COUNT(*) AS labelCount,
        MAX(MATCH(searchable_name, indications) AGAINST(? IN NATURAL LANGUAGE MODE)) AS relevance
      FROM drug_labels
      WHERE MATCH(searchable_name, indications) AGAINST(? IN NATURAL LANGUAGE MODE)
    `
    const bindings: unknown[] = [normalized, normalized]

    if (productType) {
      sql += ' AND product_type = ?'
      bindings.push(productType)
    }

    sql += `
      GROUP BY brand_name, generic_name
      ORDER BY relevance DESC
      LIMIT ? OFFSET ?
    `
    bindings.push(limit, offset)

    const rows = await db.rawQuery(sql, bindings)
    return this.mapSearchRows(rows[0])
  }

  /**
   * LIKE indication-scope fallback (query < 3 chars or FULLTEXT unavailable).
   *
   * Searches searchable_name OR indications so short queries still return
   * useful results without requiring the FULLTEXT index.
   */
  private async searchIndicationLike(
    normalized: string,
    productType: string | undefined,
    limit: number,
    offset: number
  ): Promise<DrugSearchResult[]> {
    const term = `%${normalized}%`
    let sql = `
      SELECT
        MIN(id) AS id,
        brand_name,
        generic_name,
        MIN(manufacturer) AS manufacturer,
        MIN(route) AS route,
        MIN(product_type) AS product_type,
        COUNT(*) AS labelCount
      FROM drug_labels
      WHERE (searchable_name LIKE ? OR indications LIKE ?)
    `
    const bindings: unknown[] = [term, term]

    if (productType) {
      sql += ' AND product_type = ?'
      bindings.push(productType)
    }

    sql += `
      GROUP BY brand_name, generic_name
      ORDER BY brand_name ASC
      LIMIT ? OFFSET ?
    `
    bindings.push(limit, offset)

    const rows = await db.rawQuery(sql, bindings)
    return this.mapSearchRows(rows[0])
  }

  private mapSearchRows(rows: any[]): DrugSearchResult[] {
    if (!Array.isArray(rows)) return []
    return rows.map((row) => ({
      id: Number(row.id),
      brand_name: row.brand_name ?? null,
      generic_name: row.generic_name ?? null,
      manufacturer: row.manufacturer ?? null,
      route: row.route ?? null,
      product_type: row.product_type ?? null,
      labelCount: Number(row.labelCount ?? row.labelcount ?? 1),
    }))
  }

  /**
   * Load the full detail for a single drug label row by its surrogate id.
   * Returns null if the row doesn't exist.
   */
  async find(id: number): Promise<DrugLabelDetail | null> {
    const { default: DrugLabel } = await import('#models/drug_label')
    const row = await DrugLabel.find(id)
    if (!row) return null

    return {
      id: row.id,
      set_id: row.set_id,
      spl_id: row.spl_id,
      version: row.version,
      brand_name: row.brand_name,
      generic_name: row.generic_name,
      manufacturer: row.manufacturer,
      product_ndc: row.product_ndc,
      route: row.route,
      product_type: row.product_type,
      indications: row.indications,
      dosage: row.dosage,
      warnings: row.warnings,
      boxed_warning: row.boxed_warning,
      drug_interactions: row.drug_interactions,
      contraindications: row.contraindications,
      when_using: row.when_using,
      stop_use: row.stop_use,
      source_updated_at: row.source_updated_at,
      ingested_at: row.ingested_at.toISO() ?? '',
    }
  }

  /**
   * Return the drug interaction text for a set of label ids.
   *
   * - Dedupes and caps the id list via parseCompareIds / MAX_COMPARE.
   * - One query: SELECT id, brand_name, generic_name, product_type,
   *   drug_interactions FROM drug_labels WHERE id IN (?).
   * - Re-orders the rows to match the requested id order so the caller's
   *   column positions are stable even if MySQL returns rows in a different
   *   order. Missing ids (non-existent in the table) are silently omitted.
   * - Returns [] for an empty or entirely-invalid id list.
   */
  async getInteractionsFor(ids: number[]): Promise<DrugInteractionEntry[]> {
    if (ids.length === 0) return []

    // Dedupe + cap (caller may already have done this, but be defensive).
    const safeIds = parseCompareIds(ids.join(',')).slice(0, MAX_COMPARE)
    if (safeIds.length === 0) return []

    const placeholders = safeIds.map(() => '?').join(', ')
    const sql = `
      SELECT id, brand_name, generic_name, product_type, drug_interactions
      FROM drug_labels
      WHERE id IN (${placeholders})
    `
    const rows = await db.rawQuery(sql, safeIds)
    const rawRows: Array<{
      id: number | string
      brand_name: string | null
      generic_name: string | null
      product_type: string | null
      drug_interactions: string | null
    }> = Array.isArray(rows[0]) ? rows[0] : []

    // Build a lookup by id so we can re-order to match the request order.
    const byId = new Map<number, DrugInteractionEntry>()
    for (const row of rawRows) {
      const entry: DrugInteractionEntry = {
        id: Number(row.id),
        brand_name: row.brand_name ?? null,
        generic_name: row.generic_name ?? null,
        product_type: row.product_type ?? null,
        drug_interactions: row.drug_interactions ?? null,
      }
      byId.set(entry.id, entry)
    }

    // Re-order: iterate safeIds, include only ids that exist in the table.
    const ordered: DrugInteractionEntry[] = []
    for (const id of safeIds) {
      const entry = byId.get(id)
      if (entry) ordered.push(entry)
    }
    return ordered
  }

  /**
   * Get current row count — what's searchable right now.
   */
  async rowCount(): Promise<number> {
    try {
      const result = await db.rawQuery('SELECT COUNT(*) AS cnt FROM drug_labels')
      const rows = result[0] as Array<{ cnt: number | string }>
      return Number(rows[0]?.cnt ?? 0)
    } catch {
      return 0
    }
  }

  /**
   * Dispatch the download phase (idempotent — deduped by deterministic jobId).
   * Auto-chains the ingest phase on completion. Returns "already running" if the
   * download job is active/waiting.
   */
  async triggerDownload() {
    return DownloadDrugDataJob.dispatch(true)
  }

  /**
   * 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 — `attemptAutoUpdate` 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,
    }
  }

  /**
   * 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
   * it fails fast with a typed result when nothing has been downloaded, rather
   * than dispatching a job that would immediately fail in the worker.
   */
  async triggerIngestFromDisk() {
    const marker = parseDownloadState(await KVStore.getValue('drugReference.downloadState'))
    if (!marker) {
      return {
        job: undefined,
        created: false,
        message: 'Nothing downloaded — run Download FDA data first.',
        nothingDownloaded: true,
      }
    }
    const result = await IngestDrugDataJob.dispatch()
    return { ...result, nothingDownloaded: false }
  }

  /**
   * Force-clear a wedged ingest and restart it from the on-disk parts.
   *
   * A worker killed mid-ingest (e.g. during a `nomad upgrade`) leaves its job
   * 'active' holding a lock BullMQ won't reclaim until lockDuration elapses, so
   * the normal dispatch refuses to start a new ingest and the UI button stays
   * disabled ("Indexing…") with no way out. Obliterating the single-purpose
   * drug-ingest queue removes the stuck job (force = even active/locked); we then
   * re-dispatch from disk. The downloaded parts and the download-state marker are
   * left untouched, so this restarts ingest WITHOUT re-downloading.
   */
  async resetAndReingest() {
    const marker = parseDownloadState(await KVStore.getValue('drugReference.downloadState'))
    if (!marker) {
      return {
        job: undefined,
        created: false,
        message: 'Nothing downloaded — run Download FDA data first.',
        nothingDownloaded: true,
      }
    }

    const queue = QueueService.getInstance().getQueue(IngestDrugDataJob.queue)
    try {
      // force: true removes the active/locked stuck job too. Scoped to the
      // single-purpose ingest queue, so nothing else is affected.
      await queue.obliterate({ force: true })
      logger.info('[DrugReferenceService] drug-ingest queue obliterated for restart')
    } catch (err) {
      logger.warn(
        `[DrugReferenceService] ingest queue obliterate failed (continuing to dispatch): ${
          err instanceof Error ? err.message : String(err)
        }`
      )
    }

    const result = await IngestDrugDataJob.dispatch()
    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
   * absent or completed (passes > 0 use auto-generated ids). Each phase has its
   * own queue + jobId, so this is called once per queue with the matching ids.
   */
  private async resolvePhaseJob(
    queueName: string,
    deterministicJobId: string
  ): Promise<Job | undefined> {
    const queueService = QueueService.getInstance()
    const queue = queueService.getQueue(queueName)

    let job = await queue.getJob(deterministicJobId)
    if (!job || (await job.getState()) === 'completed') {
      const activeJobs = await queue.getJobs(['active', 'waiting', 'delayed'])
      const continuation = activeJobs
        .filter((j) => j.id !== deterministicJobId)
        .sort((a, b) => (b.data?.partIndex ?? 0) - (a.data?.partIndex ?? 0))[0]
      if (continuation) {
        job = continuation
      } else {
        // No live continuation. If the chain FAILED partway, surface the failed
        // job so the status reads 'failed' instead of falsely 'ready' with only
        // part 1's count. BUT removeOnFail keeps history: a STALE failure from a
        // prior run must not mask a newer successful one (seen live: a fixed
        // re-run completed, yet the panel stayed 'failed' on last night's
        // duplicate-key job). Compare finish times and only surface the failed
        // job when it is the most recent terminal outcome.
        const [failedJobs, completedJobs] = await Promise.all([
          queue.getJobs(['failed']),
          queue.getJobs(['completed']),
        ])
        const newestFailed = failedJobs.sort((a, b) => (b.finishedOn ?? 0) - (a.finishedOn ?? 0))[0]
        const newestCompletedFinish = completedJobs.reduce(
          (max, j) => Math.max(max, j.finishedOn ?? 0),
          0
        )
        if (newestFailed && (newestFailed.finishedOn ?? 0) > newestCompletedFinish) {
          job = newestFailed
        }
      }
    }
    return job
  }

  /** Map a BullMQ job state to the per-phase run state. */
  private phaseStateFor(state: string | undefined): DrugPhaseState {
    if (state === 'failed') return 'failed'
    if (state === 'completed') return 'completed'
    if (state === 'active' || state === 'waiting' || state === 'delayed') return 'running'
    return 'idle'
  }

  /**
   * Return the two-phase ingest status for the UI panel. Reads the download job
   * (drug-download queue) and the ingest job (drug-ingest queue) independently,
   * merges the KV download-state marker + last-updated marker + live row count,
   * and derives the top-level phase from the two sub-phases.
   */
  async getIngestStatus(): Promise<DrugIngestStatus> {
    const [downloadJob, ingestJob] = await Promise.all([
      this.resolvePhaseJob(DownloadDrugDataJob.queue, DownloadDrugDataJob.jobId),
      this.resolvePhaseJob(IngestDrugDataJob.queue, IngestDrugDataJob.jobId),
    ])

    const [lastUpdated, rawMarker, count] = await Promise.all([
      KVStore.getValue('drugReference.lastUpdatedExportDate'),
      KVStore.getValue('drugReference.downloadState'),
      this.rowCount(),
    ])
    const marker = parseDownloadState(rawMarker)

    // ── Download sub-status ─────────────────────────────────────────────────
    const dlState = downloadJob ? await downloadJob.getState() : undefined
    const dlData = downloadJob?.data ?? {}
    let downloadPhaseState = this.phaseStateFor(dlState)
    // A finished download job is pruned (removeOnComplete) but the marker proves
    // the parts are on disk — treat that as a completed download phase so the
    // manual "Ingest into search" button stays available.
    if (downloadPhaseState === 'idle' && marker) downloadPhaseState = 'completed'

    const download: DrugDownloadStatus = {
      state: downloadPhaseState,
      partsDone: downloadPhaseState === 'completed' ? (marker?.totalParts ?? dlData.totalParts ?? 0) : (dlData.partIndex ?? 0),
      totalParts: dlData.totalParts ?? marker?.totalParts ?? 0,
      bytesDownloaded: dlData.bytesDownloaded,
      currentPartName: dlData.currentPartName ?? null,
      failedReason:
        dlState === 'failed' ? summarizeJobError(downloadJob?.failedReason) : undefined,
    }

    // ── Ingest sub-status ───────────────────────────────────────────────────
    const ingState = ingestJob ? await ingestJob.getState() : undefined
    const ingData = ingestJob?.data ?? {}
    let ingestPhaseState = this.phaseStateFor(ingState)
    // A pruned-but-successful ingest leaves rows + the last-updated marker and
    // clears the download marker; reflect that as a completed ingest phase.
    if (ingestPhaseState === 'idle' && !marker && count > 0) ingestPhaseState = 'completed'

    // total_records 0 means "unknown" (e.g. a manifest rebuilt from an older
    // marker) — resolveExpectedTotal falls back to a parts estimate, then the
    // live row count, so the counter/%/ETA never silently vanish.
    const expectedTotal = resolveExpectedTotal(
      ingData.manifest?.total_records,
      ingData.totalParts,
      count
    )
    const jobRecords = ingData.recordsIngested ?? 0
    // While ingesting, the per-job recordsIngested lags across the per-part
    // continuation handoff (continuations run under auto jobIds). Drive the shown
    // count from max(jobRecords, live rowCount) so a first ingest tracks the table
    // filling 0 → ~259k, while a re-ingest into a populated table still rides the
    // per-job counter. Outside the running phase, trust the job's own total.
    // Always reconcile against the live row count (the per-job recordsIngested can
    // be a partial/stale total — a completed pass-0 job only counted part 1; a
    // failed continuation stops mid-run). WHILE RUNNING, additionally subtract the
    // run's start-row baseline so a re-ingest into a populated table shows THIS
    // run's progress (0 → ~259k) instead of reading ~100% from second zero.
    // Outside running, start=0 so 'ready'/'failed' reflect total searchable rows.
    const records =
      ingestPhaseState === 'running'
        ? resolveIngestRecordsShown(jobRecords, count, expectedTotal, ingData.startRowCount ?? 0)
        : resolveIngestRecordsShown(jobRecords, count, expectedTotal)

    const ingest: DrugIngestPhaseStatus = {
      state: ingestPhaseState,
      records,
      expectedTotal,
      partsDone: ingData.partIndex ?? 0,
      totalParts: ingData.totalParts ?? marker?.totalParts ?? 0,
      currentPartName: ingData.currentPartName ?? null,
      failedReason: ingState === 'failed' ? summarizeJobError(ingestJob?.failedReason) : undefined,
    }

    const phase = deriveIngestPhase(download, ingest, count)

    // The active phase drives elapsed/ETA: ingest start when ingesting, else the
    // download start.
    const startedAtMs =
      phase === 'ingesting'
        ? (ingData.startedAt ?? null)
        : phase === 'downloading'
          ? (dlData.startedAt ?? null)
          : null

    const error =
      phase === 'failed' ? (ingest.failedReason ?? download.failedReason) : undefined

    return {
      phase,
      download,
      ingest,
      startedAtMs,
      lastUpdated: lastUpdated ?? null,
      rowCount: count,
      error,
    }
  }
}