<?php
/**
 * CollectionEngine — opinionated pipeline for the Collection Dashboard.
 *
 * Three independent ingest flows. Each writes to a *_stage table first and
 * only swaps into the live table on success, so the live data is never empty
 * between the moment an upload starts and the moment it finishes.
 *
 *   $eng = new CollectionEngine();
 *   $eng->ingestCollection('/tmp/Collection Raw.csv', $userId, $reportId);
 *   $eng->ingestFacilityMapping('/tmp/Facility Mapping.csv', $userId, $reportId);
 *
 * After ingest the engine derives, per report:
 *   - dos_year / dos_month from DOS
 *   - facility from facility_mappings (visit_id lookup)
 *   - expected_payment = per-CPT average primary-insurance payment per unit × units
 *
 * NOTE: The dashboard reports balances using the source columns
 *       `insurance_balance` and `patient_balance` (parsed directly from the CSV).
 *       The derived `outstanding` column is kept in the schema for safe rollback
 *       but is no longer written to or read from.
 */

require_once __DIR__ . '/xlsx-reader.php';

/**
 * Claim-status filter — shared by the dashboard (admin/collection.php) and the
 * drill endpoint (admin/collection-drill.php) so both apply the SAME definition.
 * The dropdown lets the ops team narrow the whole dashboard to a claim subset
 * (recently billed, unpaid, primary/secondary pending, …) — every tab, table,
 * drill-down and export then operates on that subset.
 */
function collection_status_options(): array
{
    return [
        ''            => 'All claims',
        'recent'      => 'Recently billed (≤30 days by Entry Date)',
        'outstanding' => 'Outstanding — any balance open',
        'unpaid'      => 'Unpaid — no primary payment',
        'prim_pend'   => 'Primary unpaid (pending, balance open)',
        'sec_pend'    => 'Secondary unpaid (pending, balance open)',
        'paid'        => 'Paid — primary has paid',
    ];
}

/** @return array{0:string,1:array} [$sqlFragment, $paramsToAppend] — empty fragment = no filter. */
function collection_status_sql(string $status, int $reportId): array
{
    switch ($status) {
        case 'recent':
            return ['entry_date IS NOT NULL AND entry_date >= DATE_SUB((SELECT MAX(entry_date) FROM collection_raw cr2 WHERE cr2.report_id = ?), INTERVAL 30 DAY)', [$reportId]];
        case 'outstanding': return ['(insurance_balance > 0 OR patient_balance > 0)', []];
        case 'unpaid':      return ['primary_ins_paid = 0', []];
        case 'prim_pend':   return ['primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0)', []];
        case 'sec_pend':    return ['primary_ins_paid > 0 AND secondary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0)', []];
        case 'paid':        return ['primary_ins_paid > 0', []];
        default:            return ['', []];
    }
}

class CollectionEngine
{
    private PDO $pdo;
    private int $batchSize = 1000;

    public function __construct()
    {
        $this->pdo = db();
        // Keep large GROUP BY / DISTINCT temp tables IN RAM (up to 256 MB) instead of spilling to the MySQL tmpdir
        // (/var/mysqltmp). On a disk-pressured server that on-disk temp file is exactly what fails with
        // "No space left on device" (Errcode 28) during the post-swap dashboard rebuild / facility re-map. The
        // in-memory ceiling is MIN(tmp_table_size, max_heap_table_size), so both are raised. Best-effort — a shared
        // host may forbid changing these per session; if so we silently carry on with the server defaults.
        try {
            $this->pdo->exec("SET SESSION tmp_table_size = 268435456, SESSION max_heap_table_size = 268435456");
        } catch (Throwable $e) { /* host disallows it — no-op */ }
    }

    /* ------------------------------------------------------------------ *
     * Public API
     * ------------------------------------------------------------------ */

    /**
     * Ingest Collection Raw for ONE report.  Replaces only that report's rows;
     * other reports' data is left untouched.
     */
    /**
     * Wipe ALL collection data + cache + upload history for one report.  Facility mapping is
     * intentionally preserved so the user doesn't have to re-upload it before the next load.
     * Cheap (a few DELETEs); fine to run inline.
     */
    public function clearReportData(int $reportId): array
    {
        $this->requireReport($reportId);
        $counts = ['live' => 0, 'stage' => 0, 'pending' => 0, 'snapshot_taken' => 0];

        // FIRST: capture a final snapshot of the current data state so the Compare tab
        // keeps a "before clear" reference.  Only computes if the latest completed upload
        // doesn't already have a snapshot in collection_stats (the usual path leaves one;
        // we cover the rare case where the deferred cache build failed last time).
        try {
            $st = $this->pdo->prepare(
                "SELECT id FROM collection_uploads
                  WHERE report_id = ? AND kind = 'collection' AND status = 'completed'
               ORDER BY id DESC LIMIT 1"
            );
            $st->execute([$reportId]);
            $latestUploadId = (int) $st->fetchColumn();
            if ($latestUploadId) {
                $hs = $this->pdo->prepare("SELECT 1 FROM collection_stats WHERE upload_id = ? LIMIT 1");
                $hs->execute([$latestUploadId]);
                if (!$hs->fetchColumn()) {
                    $this->computeAndCacheStats($latestUploadId, $reportId);   // snapshot the current live data
                    $counts['snapshot_taken'] = 1;
                }
            }
        } catch (Throwable $t) {
            log_error_db('Clear: pre-wipe snapshot capture failed (continuing): ' . $t->getMessage(),
                         'clear', ['report_id' => $reportId], 'warning');
        }

        // CHUNKED delete — a single `DELETE FROM collection_raw WHERE report_id = ?` on
        // 200k+ rows with 5 wide covering indexes overwhelms a CPU-throttled shared host
        // and drops the connection (2006). Each chunk is a tiny autocommit transaction;
        // the connection stays warm and is reconnected automatically if it ever drops.
        $chunkDelete = function (string $table, string $where, array $params, int $chunk = 5000): int {
            $sql   = "DELETE FROM `$table` WHERE $where LIMIT $chunk";
            $total = 0;
            $stuck = 0;                          // CONSECUTIVE lock failures on the current chunk
            while (true) {
                try {
                    $st = $this->pdo->prepare($sql);
                    $st->execute($params);
                    $n = $st->rowCount();
                    $total += $n;
                    $stuck = 0;                  // progress made — reset the budget for the next chunk
                    if ($n < $chunk) break;     // last (partial) chunk — done
                } catch (Throwable $t) {
                    if ($this->isGoneAway($t)) { $this->reconnect(); continue; }   // retry same chunk on fresh conn
                    /* Lock contention is TRANSIENT: back off briefly and retry the same chunk rather than
                     * abandoning the whole wipe. Counted per-chunk and reset on success, so a long delete that
                     * hits occasional contention keeps going, while a genuinely stuck row (someone holding a
                     * lock indefinitely) still surfaces as a real error instead of looping forever. */
                    if ($this->isLockContention($t) && $stuck < 8) {
                        $stuck++;
                        usleep(250000 * $stuck);        // 0.25s, 0.5s, 0.75s … ~9s total before giving up
                        continue;
                    }
                    throw $t;
                }
            }
            return $total;
        };

        // Heavy tables — chunked.  Only the LIVE working data is wiped.
        $counts['live']  = $chunkDelete('collection_raw',       'report_id = ?', [$reportId], 5000);
        $counts['stage'] = $chunkDelete('collection_raw_stage', 'report_id = ?', [$reportId], 5000);
        try { $counts['pending'] = $chunkDelete('pending_uploads', 'report_id = ?', [$reportId], 1000); }
        catch (Throwable $e) { /* table may not exist on a brand-new deploy */ }

        // INTENTIONALLY PRESERVED across Clear:
        //   • collection_stats  → snapshot history (Compare tab survives clears)
        //   • collection_uploads → audit / history trail per upload
        //   • facility_mappings  → mapping persists (user explicitly wants this)
        return $counts;
    }

    /** Clear ALL facility-mapping data for a report (the facility_mappings table + its stage). Mirrors
     *  clearReportData's chunked, reconnect-safe delete. Upload-history rows (collection_uploads) are kept for the
     *  audit trail; already-ingested collection_raw rows keep their baked-in facility (mapping is applied at ingest). */
    public function clearFacilityMapping(int $reportId): array
    {
        $this->requireReport($reportId);
        $counts = ['deleted' => 0, 'stage' => 0];
        $chunkDelete = function (string $table, string $where, array $params, int $chunk = 5000): int {
            $sql = "DELETE FROM `$table` WHERE $where LIMIT $chunk"; $total = 0; $stuck = 0;
            while (true) {
                try {
                    $st = $this->pdo->prepare($sql); $st->execute($params); $n = $st->rowCount(); $total += $n;
                    $stuck = 0;
                    if ($n < $chunk) break;
                } catch (Throwable $t) {
                    if ($this->isGoneAway($t)) { $this->reconnect(); continue; }
                    // Same transient-lock retry as clearReportData — see the note there.
                    if ($this->isLockContention($t) && $stuck < 8) { $stuck++; usleep(250000 * $stuck); continue; }
                    throw $t;
                }
            }
            return $total;
        };
        $counts['deleted'] = $chunkDelete('facility_mappings', 'report_id = ?', [$reportId], 5000);
        try { $counts['stage'] = $chunkDelete('facility_mappings_stage', 'report_id = ?', [$reportId], 5000); }
        catch (Throwable $e) { /* stage table may be empty / absent on a fresh deploy */ }
        // Invalidate the cached KPI counts on the completed upload rows so the page reflects the now-EMPTY mapping
        // (the tiles read meta_json from the last completed upload; without this they show stale pre-clear counts).
        try { $this->pdo->prepare("UPDATE collection_uploads SET meta_json=NULL WHERE report_id=? AND kind='facility_mapping' AND status='completed'")->execute([$reportId]); }
        catch (Throwable $e) {}
        return $counts;
    }

    /** Clear ALL deductible-mapping data for a report. Unlike facility (whose baked-in name persists), the
     *  deductible flag is a boolean set defined ENTIRELY by the uploaded list, so clearing it un-flags the
     *  collection rows too (deductible = 0). Upload-history rows are kept for the audit trail. */
    public function clearDeductibleMapping(int $reportId): array
    {
        $this->requireReport($reportId);
        $counts = ['deleted' => 0, 'stage' => 0];
        $chunkDelete = function (string $table, string $where, array $params, int $chunk = 5000): int {
            $sql = "DELETE FROM `$table` WHERE $where LIMIT $chunk"; $total = 0; $stuck = 0;
            while (true) {
                try {
                    $st = $this->pdo->prepare($sql); $st->execute($params); $n = $st->rowCount(); $total += $n;
                    $stuck = 0;
                    if ($n < $chunk) break;
                } catch (Throwable $t) {
                    if ($this->isGoneAway($t)) { $this->reconnect(); continue; }
                    // Same transient-lock retry as clearReportData — see the note there.
                    if ($this->isLockContention($t) && $stuck < 8) { $stuck++; usleep(250000 * $stuck); continue; }
                    throw $t;
                }
            }
            return $total;
        };
        $counts['deleted'] = $chunkDelete('deductible_mappings', 'report_id = ?', [$reportId], 5000);
        try { $counts['stage'] = $chunkDelete('deductible_mappings_stage', 'report_id = ?', [$reportId], 5000); }
        catch (Throwable $e) { /* stage table may be empty / absent on a fresh deploy */ }
        // Un-flag the collection rows (only the currently-flagged ones → cheap; avoids rewriting the report).
        try { $this->pdo->prepare("UPDATE collection_raw SET deductible = 0 WHERE report_id = ? AND deductible <> 0")->execute([$reportId]); }
        catch (Throwable $e) {}
        try { $this->pdo->prepare("UPDATE collection_uploads SET meta_json=NULL WHERE report_id=? AND kind='deductible_mapping' AND status='completed'")->execute([$reportId]); }
        catch (Throwable $e) {}
        return $counts;
    }

    public function ingestCollection(string $filePath, int $userId, int $reportId, bool $appendMode = false): array
    {
        $this->requireReport($reportId);
        $uploadId = $this->createUploadRecord('collection', $filePath, $userId, null, $reportId);

        try {
            // 1.  Clear staging rows for this report only
            $this->pdo->prepare('DELETE FROM collection_raw_stage WHERE report_id = ?')->execute([$reportId]);

            // 2.  Stream rows into stage tagged with report_id.  This now ALSO pre-derives
            //     dos_year/month, days-to-first-payment, facility, entry_date and upload_id
            //     in PHP per row — those used to be 2 full-table UPDATEs over every row.
            [$total, $valid, $errs] = $this->loadCollectionToStage($filePath, $reportId, $uploadId);
            if ($valid === 0) {
                throw new RuntimeException('No valid rows found in file. Check the column headers.');
            }

            // 3.  Only ONE derivation UPDATE left — expected_payment (it needs the
            //     full-file aggregate per CPT, so it can't be done row-by-row in PHP).
            //     `outstanding` is no longer derived; the dashboard reads insurance_balance
            //     and patient_balance directly from the CSV-loaded source columns.
            $this->derivedCollectionOnStage($reportId);

            // 4.  Atomic swap (REPLACE mode) — delete THIS report's rows from live, then INSERT
            //     from stage.  In APPEND mode the DELETE is skipped — new rows are simply added
            //     to whatever's already there.  upload_id was already stamped on stage rows
            //     (pass 1), so no extra UPDATE is needed.  Wrapped with reconnect-on-gone-away.
            $swap = function () use ($reportId, $appendMode) {
                $this->safeBegin();
                if (!$appendMode) {
                    $this->pdo->prepare('DELETE FROM collection_raw WHERE report_id = ?')->execute([$reportId]);
                }
                $this->pdo->prepare("INSERT INTO collection_raw SELECT * FROM collection_raw_stage WHERE report_id = ?")
                          ->execute([$reportId]);
                $this->safeCommit();
            };
            try {
                $swap();
            } catch (Throwable $sw) {
                if ($this->isGoneAway($sw)) {
                    try { if ($this->pdo->inTransaction()) $this->pdo->rollBack(); } catch (Throwable $r) {}
                    log_error_db('Swap dropped the connection; reconnecting and retrying once: ' . $sw->getMessage(),
                                 'ingest', ['upload_id' => $uploadId, 'report_id' => $reportId, 'append' => $appendMode], 'warning');
                    $this->reconnect();
                    $swap();
                } else {
                    throw $sw;
                }
            }

            // Stage rows are no longer needed once the swap has committed — prune them so collection_raw_stage stops
            // holding a full stale copy of every upload (keeps the DB footprint down; frees room for the next run).
            // Best-effort: a failure here must NOT fail the (already-live) upload. If it drops the connection, reconnect
            // so the following completeUploadRecord()/cache steps don't run on a dead handle and mislabel the upload.
            try { $this->pdo->prepare('DELETE FROM collection_raw_stage WHERE report_id = ?')->execute([$reportId]); }
            catch (Throwable $e) { if ($this->isGoneAway($e)) { try { $this->reconnect(); } catch (Throwable $r) {} } }

            // APPEND mode only — the per-CPT rate computed during pass 3 was based on THIS
            // batch's rows alone, so older rows' expected_payment may be stale relative to
            // the now-combined data.  Recompute across the full report so all rows agree
            // on a single rate.  (REPLACE mode doesn't need this — the swap already replaced
            // everything with the freshly-derived stage rows.)
            // NOTE: `outstanding` is no longer derived — insurance_balance and patient_balance
            // come straight from the CSV per row, so no full-report recompute is needed for them.
            if ($appendMode) {
                try {
                    $this->pdo->prepare("
                        UPDATE collection_raw cr
                        LEFT JOIN (
                            SELECT charge_code,
                                   SUM(primary_ins_paid)
                                     / NULLIF(SUM(CASE WHEN primary_ins_paid > 0 THEN COALESCE(units,1) ELSE 0 END), 0) AS rate
                            FROM collection_raw
                            WHERE report_id = ? AND primary_ins_paid > 0
                              AND charge_code IS NOT NULL AND charge_code <> ''
                            GROUP BY charge_code
                        ) r ON r.charge_code = cr.charge_code
                        SET cr.expected_payment = ROUND(COALESCE(r.rate, 0) * COALESCE(cr.units, 1), 2)
                        WHERE cr.report_id = ?
                    ")->execute([$reportId, $reportId]);
                } catch (Throwable $rc) {
                    log_error_db('Append: post-append expected_payment recompute failed: ' . $rc->getMessage(),
                                 'ingest', ['report_id' => $reportId, 'upload_id' => $uploadId], 'warning');
                }
            }

            // The new data is LIVE and correct now → mark the upload completed immediately
            // so the uploader page stops waiting.  The dashboard analytics cache is then
            // rebuilt AFTER, off the critical path; the dashboard falls back to live SQL
            // for any view whose cache isn't ready yet, so nothing breaks in the meantime.
            $this->completeUploadRecord($uploadId, $total, $valid, $errs);
            try {
                $this->computeAndCacheStats($uploadId, $reportId);   // dashboard cache (best-effort)
            } catch (Throwable $e) {
                log_error_db('computeAndCacheStats (post-complete) failed: ' . $e->getMessage(), 'ingest', ['upload_id' => $uploadId, 'report_id' => $reportId], 'warning');
            }
            // No mirror sync here — cross-report facility rollups (e.g. 12 Blue Central
            // pulling from reports 7/8/10) are applied VIRTUALLY at query time by
            // `collection_scope_for_report()` in admin/collection.php and admin/collection-drill.php.
            // Source ingests automatically flow into rollup views on next page load without any
            // row-copying or extra DELETE/INSERT pass here.
            return ['upload_id'=>$uploadId,'total'=>$total,'valid'=>$valid,'errors'=>$errs];
        } catch (Throwable $t) {
            // Recovery: the original $t may have killed the connection (2006), in which case
            // rollBack / failUploadRecord will throw AGAIN and we'd lose the real error message.
            // Wrap each step, reconnect once on a dropped connection, so the failed upload
            // still gets recorded (status='failed' + note) and $t is re-thrown intact.
            try { if ($this->pdo->inTransaction()) $this->pdo->rollBack(); } catch (Throwable $r) {}
            try {
                $this->failUploadRecord($uploadId, $t->getMessage());
            } catch (Throwable $f) {
                if ($this->isGoneAway($f) || $this->isGoneAway($t)) {
                    try { $this->reconnect(); $this->failUploadRecord($uploadId, $t->getMessage()); }
                    catch (Throwable $f2) { /* give up — the original $t is logged by the caller */ }
                }
            }
            throw $t;
        }
    }

    public function ingestFacilityMapping(string $filePath, int $userId, int $reportId, bool $skipCollectionRefresh = false): array
    {
        $this->requireReport($reportId);
        $uploadId = $this->createUploadRecord('facility_mapping', $filePath, $userId, null, $reportId);

        try {
            $this->pdo->prepare('DELETE FROM facility_mappings_stage WHERE report_id = ?')->execute([$reportId]);
            [$total, $valid, $errs] = $this->loadFacilityToStage($filePath, $reportId);
            if ($valid === 0) throw new RuntimeException('No valid rows. Expected columns: "Visit Number" and "Facility".');

            $this->safeBegin();
            $this->pdo->prepare('DELETE FROM facility_mappings WHERE report_id = ?')->execute([$reportId]);
            $this->pdo->prepare("INSERT INTO facility_mappings SELECT * FROM facility_mappings_stage WHERE report_id = ?")
                      ->execute([$reportId]);
            $this->pdo->prepare("UPDATE facility_mappings SET upload_id = ? WHERE report_id = ? AND upload_id IS NULL")
                      ->execute([$uploadId, $reportId]);
            $this->safeCommit();

            // Stage rows are no longer needed once the swap has committed — prune them so facility_mappings_stage
            // stops holding a full stale copy of every upload. Best-effort (never fail the already-live upload);
            // reconnect on a dropped connection so the following cache/complete steps don't run on a dead handle.
            try { $this->pdo->prepare('DELETE FROM facility_mappings_stage WHERE report_id = ?')->execute([$reportId]); }
            catch (Throwable $e) { if ($this->isGoneAway($e)) { try { $this->reconnect(); } catch (Throwable $r) {} } }

            // Cache the mapping KPI counts (unique visits / facilities) on the
            // upload row so the Facility Mapping page never has to scan the
            // whole table on load.  Computed once here, read O(1) per page view.
            $this->cacheMappingMeta($uploadId, $reportId);

            // Re-derive facility names on this report's collection rows + rebuild the
            // dashboard cache.  SKIPPED when a Collection import follows immediately
            // (combined upload) — that import re-derives facility and rebuilds the
            // cache once, so we don't pay for the expensive recompute twice.
            if (!$skipCollectionRefresh) {
                // BEST-EFFORT: the mapping swap has ALREADY committed (the mapping is live), so a failure of the heavy
                // collection re-map / dashboard-cache rebuild — which builds large temp tables over a big collection_raw
                // and is exactly what fails with "No space left on device" on a disk-pressured server — must NOT fail the
                // whole upload. If it throws, the mapping still uploaded; the dashboard re-derives on its next rebuild.
                try {
                    $this->applyFacilityToCollection($reportId);
                    $liveUploadId = (int) $this->latestUploadId('collection', $reportId);
                    if ($liveUploadId) $this->computeAndCacheStats($liveUploadId, $reportId);
                    // No mirror sync — rollups are applied virtually at query time, so a facility-mapping change on a
                    // source report becomes visible in every rollup view automatically on the next dashboard load.
                } catch (Throwable $e) {
                    if ($this->isGoneAway($e)) { try { $this->reconnect(); } catch (Throwable $r) {} }
                    try { log_error_db('Facility upload: collection re-map/cache deferred (mapping is live): ' . $e->getMessage(), 'ingest', ['report_id' => $reportId, 'upload_id' => $uploadId], 'warning'); } catch (Throwable $r) {}
                }
            }

            $this->completeUploadRecord($uploadId, $total, $valid, $errs);
            return ['upload_id'=>$uploadId,'total'=>$total,'valid'=>$valid,'errors'=>$errs];
        } catch (Throwable $t) {
            try { if ($this->pdo->inTransaction()) $this->pdo->rollBack(); } catch (Throwable $r) {}
            try {
                $this->failUploadRecord($uploadId, $t->getMessage());
            } catch (Throwable $f) {
                if ($this->isGoneAway($f) || $this->isGoneAway($t)) {
                    try { $this->reconnect(); $this->failUploadRecord($uploadId, $t->getMessage()); }
                    catch (Throwable $f2) {}
                }
            }
            throw $t;
        }
    }

    /** Ingest a Deductible visit-id file (mirror of ingestFacilityMapping). The upload carries ONLY the
     *  deductible Visit Numbers; the flag is baked onto collection_raw.deductible (charge is read from
     *  collection_raw at query time). Atomic per-report stage→live swap, then re-map + cache rebuild. */
    public function ingestDeductibleMapping(string $filePath, int $userId, int $reportId, bool $skipCollectionRefresh = false): array
    {
        $this->requireReport($reportId);
        $uploadId = $this->createUploadRecord('deductible_mapping', $filePath, $userId, null, $reportId);

        try {
            $this->pdo->prepare('DELETE FROM deductible_mappings_stage WHERE report_id = ?')->execute([$reportId]);
            [$total, $valid, $errs] = $this->loadDeductibleToStage($filePath, $reportId);
            if ($valid === 0) throw new RuntimeException('No valid rows. Expected a "Visit Number" column.');

            $this->safeBegin();
            $this->pdo->prepare('DELETE FROM deductible_mappings WHERE report_id = ?')->execute([$reportId]);
            $this->pdo->prepare("INSERT INTO deductible_mappings SELECT * FROM deductible_mappings_stage WHERE report_id = ?")
                      ->execute([$reportId]);
            $this->pdo->prepare("UPDATE deductible_mappings SET upload_id = ? WHERE report_id = ? AND upload_id IS NULL")
                      ->execute([$uploadId, $reportId]);
            $this->safeCommit();

            try { $this->pdo->prepare('DELETE FROM deductible_mappings_stage WHERE report_id = ?')->execute([$reportId]); }
            catch (Throwable $e) { if ($this->isGoneAway($e)) { try { $this->reconnect(); } catch (Throwable $r) {} } }

            $this->cacheDeductibleMeta($uploadId, $reportId);

            // Re-flag collection rows + rebuild the dashboard cache. BEST-EFFORT (the mapping swap is already
            // committed / live) so a heavy re-map/cache rebuild that trips the temp-limited box never fails the
            // upload — the dashboard re-derives on its next rebuild. SKIPPED for a combined upload.
            if (!$skipCollectionRefresh) {
                try {
                    $this->applyDeductibleToCollection($reportId);
                    $liveUploadId = (int) $this->latestUploadId('collection', $reportId);
                    if ($liveUploadId) $this->computeAndCacheStats($liveUploadId, $reportId);
                } catch (Throwable $e) {
                    if ($this->isGoneAway($e)) { try { $this->reconnect(); } catch (Throwable $r) {} }
                    try { log_error_db('Deductible upload: collection re-map/cache deferred (mapping is live): ' . $e->getMessage(), 'ingest', ['report_id' => $reportId, 'upload_id' => $uploadId], 'warning'); } catch (Throwable $r) {}
                }
            }

            $this->completeUploadRecord($uploadId, $total, $valid, $errs);
            return ['upload_id'=>$uploadId,'total'=>$total,'valid'=>$valid,'errors'=>$errs];
        } catch (Throwable $t) {
            try { if ($this->pdo->inTransaction()) $this->pdo->rollBack(); } catch (Throwable $r) {}
            try {
                $this->failUploadRecord($uploadId, $t->getMessage());
            } catch (Throwable $f) {
                if ($this->isGoneAway($f) || $this->isGoneAway($t)) {
                    try { $this->reconnect(); $this->failUploadRecord($uploadId, $t->getMessage()); }
                    catch (Throwable $f2) {}
                }
            }
            throw $t;
        }
    }

    /* ------------------------------------------------------------------ *
     * Multi-tenant helpers
     * ------------------------------------------------------------------ */

    private function requireReport(int $reportId): void
    {
        if ($reportId <= 0) {
            throw new RuntimeException('A report_id is required.  Open the Collection Dashboard for this report and use its Replace Collection / Mapping buttons.');
        }
        $st = $this->pdo->prepare("SELECT id FROM reports WHERE id = ? AND deleted_at IS NULL LIMIT 1");
        $st->execute([$reportId]);
        if (!$st->fetchColumn()) throw new RuntimeException("Report #$reportId not found or deleted.");
    }

    private function latestUploadId(string $kind, int $reportId): int
    {
        $st = $this->pdo->prepare("SELECT id FROM collection_uploads
                                    WHERE kind = ? AND status = 'completed' AND report_id = ?
                                    ORDER BY completed_at DESC LIMIT 1");
        $st->execute([$kind, $reportId]);
        return (int) ($st->fetchColumn() ?: 0);
    }

    /**
     * Compute the Facility Mapping page's KPI counts once (at ingest) and store
     * them as JSON on the upload row.  COUNT(DISTINCT facility) over ~800k rows
     * with the wrong index took 134s; here we force the covering indexes so it
     * runs in well under a second, and the page then reads the cached numbers
     * instead of scanning the table on every load.
     */
    private function cacheMappingMeta(int $uploadId, int $reportId): void
    {
        try {
            $st = $this->pdo->prepare("SELECT COUNT(DISTINCT visit_id)
                                         FROM facility_mappings USE INDEX(idx_fac_map_report_visit)
                                        WHERE report_id = ?");
            $st->execute([$reportId]);
            $uniqVisits = (int) $st->fetchColumn();

            $st = $this->pdo->prepare("SELECT COUNT(DISTINCT facility)
                                         FROM facility_mappings USE INDEX(idx_fac_map_report_facility)
                                        WHERE report_id = ? AND facility IS NOT NULL AND facility <> ''");
            $st->execute([$reportId]);
            $uniqFacilities = (int) $st->fetchColumn();

            $meta = json_encode([
                'uniq_visits'     => $uniqVisits,
                'uniq_facilities' => $uniqFacilities,
                'computed_at'     => date('c'),
            ]);
            $this->pdo->prepare("UPDATE collection_uploads SET meta_json = ? WHERE id = ?")
                      ->execute([$meta, $uploadId]);
        } catch (Throwable $t) {
            // Non-fatal: the page has a live fallback if meta_json is absent.
        }
    }

    /** Cache the Deductible page's KPI count (unique deductible visits) once at ingest (mirror of cacheMappingMeta). */
    private function cacheDeductibleMeta(int $uploadId, int $reportId): void
    {
        try {
            $st = $this->pdo->prepare("SELECT COUNT(DISTINCT visit_id)
                                         FROM deductible_mappings USE INDEX(idx_ded_map_report_visit)
                                        WHERE report_id = ?");
            $st->execute([$reportId]);
            $uniqVisits = (int) $st->fetchColumn();
            $meta = json_encode(['uniq_visits' => $uniqVisits, 'computed_at' => date('c')]);
            $this->pdo->prepare("UPDATE collection_uploads SET meta_json = ? WHERE id = ?")->execute([$meta, $uploadId]);
        } catch (Throwable $t) {
            // Non-fatal: the page has a live fallback if meta_json is absent.
        }
    }

    /* ------------------------------------------------------------------ *
     * Stage loaders
     * ------------------------------------------------------------------ */

    private function iterateCsv(string $filePath, string $delim = ','): Generator
    {
        $fh = fopen($filePath, 'r');
        if (!$fh) throw new RuntimeException("Cannot open $filePath");
        try {
            // Skip BOM if present
            $first = fgets($fh);
            if ($first === false) return;
            $first = preg_replace('/^\xEF\xBB\xBF/', '', $first);
            rewind($fh);
            // If we skipped BOM, re-read first line cleanly
            $headerLine = fgetcsv($fh, 0, $delim);
            if ($headerLine && isset($headerLine[0])) {
                $headerLine[0] = preg_replace('/^\xEF\xBB\xBF/', '', $headerLine[0]);
            }
            yield 0 => $headerLine;
            $rn = 0;
            while (($row = fgetcsv($fh, 0, $delim)) !== false) {
                $rn++;
                yield $rn => $row;
            }
        } finally {
            fclose($fh);
        }
    }

    private function iterateAny(string $filePath): Generator
    {
        $ext = strtolower(pathinfo($filePath, PATHINFO_EXTENSION));
        if ($ext === 'csv' || $ext === 'tsv') {
            return $this->iterateCsv($filePath, $ext === 'tsv' ? "\t" : ',');
        }
        if ($ext === 'xlsx') {
            $reader = new XlsxReader($filePath);
            return $reader->rows();
        }
        throw new RuntimeException("Unsupported file extension: $ext");
    }

    private function loadCollectionToStage(string $filePath, int $reportId, int $uploadId): array
    {
        // Header → DB column map (case + whitespace insensitive)
        $map = [
            'provider name'                  => 'provider_name',
            'chart number'                   => 'chart_number',
            'visitid'                        => 'visit_id',
            'visit id'                       => 'visit_id',
            'patient last name'              => 'patient_last_name',
            'patient first name'             => 'patient_first_name',
            'financial class'                => 'financial_class',
            'carrier 1 code'                 => 'carrier1_code',
            'carrier 1 name'                 => 'carrier1_name',
            'carrier 2 code'                 => 'carrier2_code',
            'carrier 2 name'                 => 'carrier2_name',
            'carrier 3 code'                 => 'carrier3_code',
            'carrier 3 name'                 => 'carrier3_name',
            'dos'                            => 'dos',
            'charge code category code'      => 'charge_code_cat',
            'charge code'                    => 'charge_code',
            'charge code descrtiption'       => 'charge_code_desc', // intentional typo to match source
            'charge code description'        => 'charge_code_desc',
            'units'                          => 'units',
            'charge amount'                  => 'charge_amount',
            'amount allowed'                 => 'amount_allowed',
            'primary insurance paid'         => 'primary_ins_paid',
            'primary insurance payment date' => 'primary_ins_payment_date',
            'secondary insurance paid'       => 'secondary_ins_paid',
            'secondary insurance payment date'=> 'secondary_ins_payment_date',
            'tertiary insurance paid'        => 'tertiary_ins_paid',
            'tertiary insurance payment date'=> 'tertiary_ins_payment_date',
            'patient payment amount'         => 'patient_payment_amount',
            'patient payment date'           => 'patient_payment_date',
            'wo amount'                      => 'wo_amount',
            'wo date'                        => 'wo_date',
            'patient balance'                => 'patient_balance',
            'insurance balance'              => 'insurance_balance',
            'diagnosis code 1'               => 'diagnosis_code1',
            'diagnosis code 1 desc'          => 'diagnosis_code1_desc',
            'diagnosis code 2'               => 'diagnosis_code2',
            'diagnosis code 2 desc'          => 'diagnosis_code2_desc',
            'diagnosis code 3'               => 'diagnosis_code3',
            'diagnosis code 3 desc'          => 'diagnosis_code3_desc',
            'diagnosis code 4'               => 'diagnosis_code4',
            'diagnosis code 4 desc'          => 'diagnosis_code4_desc',
            'patient zip code'               => 'patient_zip',
            'referring provider'             => 'referring_provider',
            'appointment date'               => 'appointment_date',
            'appointment types'              => 'appointment_types',
        ];

        $iter = $this->iterateCsv($filePath);
        $iter->rewind();
        $header = $iter->current();
        if (!$header) throw new RuntimeException('Empty file.');

        // Build idx -> column
        $idxToCol = [];
        foreach ($header as $i => $h) {
            $k = strtolower(trim((string) $h));
            if (isset($map[$k])) $idxToCol[$i] = $map[$k];
        }
        if (empty($idxToCol)) throw new RuntimeException('No recognizable column headers. First row: ' . implode(',', array_slice($header, 0, 10)));
        if (!in_array('visit_id', $idxToCol, true)) throw new RuntimeException('Required column "VisitID" not found in header.');

        // Build the per-row column list: report_id + mapped CSV columns + 6 PRE-DERIVED columns
        // we used to compute with separate full-table UPDATEs (dos_year/month, days-to-pay,
        // facility, entry_date, upload_id).  Doing this in PHP at load time avoids rewriting
        // every row twice more — a big win for large files on a CPU-throttled host.
        $cols     = array_values(array_unique($idxToCol));
        $allCols  = array_merge(['report_id'], $cols, ['dos_year', 'dos_month', 'days_to_first_payment', 'facility', 'entry_date', 'deductible', 'upload_id']);
        $visitPos = array_search('visit_id', $cols, true);
        $dosPos   = array_search('dos', $cols, true);
        $ppPos    = array_search('primary_ins_payment_date', $cols, true);

        // Load THIS report's facility mapping into memory once (visit_id → [facility, entry_date])
        // so the per-row facility/entry_date lookup is a hash hit instead of a JOIN UPDATE later.
        $famap = [];
        try {
            $fs = $this->pdo->prepare("SELECT visit_id, MAX(facility) AS facility, MAX(entry_date) AS entry_date
                                         FROM facility_mappings
                                        WHERE report_id = ? AND facility IS NOT NULL AND facility <> ''
                                        GROUP BY visit_id");
            $fs->execute([$reportId]);
            while ($fr = $fs->fetch()) { $famap[(string) $fr['visit_id']] = [$fr['facility'], $fr['entry_date']]; }
        } catch (Throwable $t) { /* mapping table absent or empty → leave facility NULL */ }

        // Load THIS report's deductible visit list into memory once (a set of visit_ids) so the per-row
        // deductible flag is a hash hit instead of a JOIN UPDATE later (mirrors $famap above).
        $dedmap = [];
        try {
            $ds = $this->pdo->prepare("SELECT DISTINCT visit_id FROM deductible_mappings WHERE report_id = ?");
            $ds->execute([$reportId]);
            while ($dr = $ds->fetch()) { $dedmap[(string) $dr['visit_id']] = 1; }
        } catch (Throwable $t) { /* deductible table absent or empty → leave deductible 0 */ }

        // Multi-row batch sized to keep the bound-param count well under PDO/MySQL limits.
        // (The LOAD DATA INFILE fast path was reverted — it triggered 2006 "server has gone
        // away" on this host. Use stable batched INSERTs; flushBatch already byte-bounds
        // each packet AND has reconnect-on-gone-away logic for transient drops.)
        $batchRows = max(1, min(1300, intdiv(60000, count($allCols))));

        $total = $valid = $errs = 0;
        $buf = []; $bufRows = 0;
        $this->safeBegin();
        $iter->next();
        while ($iter->valid()) {
            $row = $iter->current();
            $iter->next();
            $total++;
            if (!is_array($row)) continue;
            $values = array_fill(0, count($cols), null);
            foreach ($idxToCol as $i => $colName) {
                $colPos = array_search($colName, $cols, true);
                $values[$colPos] = $this->normalizeCellForCollection($colName, $row[$i] ?? null);
            }
            if ($visitPos === false || $values[$visitPos] === null || $values[$visitPos] === '') { $errs++; continue; }

            // Pre-derive in PHP (replaces 2 of the 3 former stage UPDATEs)
            $dosVal = ($dosPos !== false) ? $values[$dosPos] : null;          // 'Y-m-d' (already normalized)
            $dosY = $dosM = null;
            if (is_string($dosVal) && strlen($dosVal) >= 7) {
                $dosY = (int) substr($dosVal, 0, 4) ?: null;
                $dosM = (int) substr($dosVal, 5, 2) ?: null;
            }
            $days = null;
            if ($dosVal && $ppPos !== false && is_string($values[$ppPos])) {
                $d1 = strtotime($dosVal); $d2 = strtotime($values[$ppPos]);
                if ($d1 && $d2) $days = (int) floor(($d2 - $d1) / 86400);
            }
            $fac = $entry = null;
            $vid = (string) $values[$visitPos];
            if (isset($famap[$vid])) { [$fac, $entry] = $famap[$vid]; }
            $ded = isset($dedmap[$vid]) ? 1 : 0;

            $buf[] = $reportId;
            foreach ($values as $v) $buf[] = $v;
            $buf[] = $dosY; $buf[] = $dosM; $buf[] = $days; $buf[] = $fac; $buf[] = $entry; $buf[] = $ded; $buf[] = $uploadId;

            if (++$bufRows >= $batchRows) {
                [$ins, $er] = $this->flushBatch('collection_raw_stage', $allCols, $buf);
                $valid += $ins; $errs += $er; $bufRows = 0;
                $this->safeCommit();
                $this->safeBegin();
            }
        }
        [$ins, $er] = $this->flushBatch('collection_raw_stage', $allCols, $buf);
        $valid += $ins; $errs += $er;
        $this->safeCommit();
        return [$total, $valid, $errs];
    }

    private function loadFacilityToStage(string $filePath, int $reportId): array
    {
        // Supporting CSV now contains ONLY two columns: Visit Number + Facility.
        // Any other columns in the header are silently ignored.
        $map = [
            'visit number' => 'visit_id',
            'visit id'     => 'visit_id',
            'visitid'      => 'visit_id',
            'visit_id'     => 'visit_id',
            'facility'     => 'facility',
            'entry date'   => 'entry_date',
            'entry_date'   => 'entry_date',
            'entrydate'    => 'entry_date',
        ];

        $iter = $this->iterateCsv($filePath);
        $iter->rewind();
        $header = $iter->current();
        if (!$header) throw new RuntimeException('Empty file.');

        $idxToCol = [];
        foreach ($header as $i => $h) {
            $k = strtolower(trim((string) $h));
            if (isset($map[$k]) && $map[$k] !== null) $idxToCol[$i] = $map[$k];
        }
        if (!in_array('visit_id', $idxToCol, true)) throw new RuntimeException('Required column "Visit Number" not found in facility mapping.');

        $cols     = array_values(array_unique($idxToCol));
        $allCols  = array_merge(['report_id'], $cols);
        $visitPos = array_search('visit_id', $cols, true);
        $batchRows = max(1, min(5000, intdiv(120000, count($allCols))));   // mapping is just 2-3 cols → very large batches (fewer round-trips)

        $total = $valid = $errs = 0;
        $buf = []; $bufRows = 0;
        $this->safeBegin();
        $iter->next();
        while ($iter->valid()) {
            $row = $iter->current();
            $iter->next();
            $total++;
            if (!is_array($row)) continue;
            $values = array_fill(0, count($cols), null);
            foreach ($idxToCol as $i => $colName) {
                $colPos = array_search($colName, $cols, true);
                $values[$colPos] = $this->normalizeCellForFacility($colName, $row[$i] ?? null);
            }
            if ($visitPos === false || $values[$visitPos] === null || $values[$visitPos] === '') { $errs++; continue; }
            $buf[] = $reportId;
            foreach ($values as $v) $buf[] = $v;
            if (++$bufRows >= $batchRows) {
                [$ins, $er] = $this->flushBatch('facility_mappings_stage', $allCols, $buf);
                $valid += $ins; $errs += $er; $bufRows = 0;
                $this->safeCommit();
                $this->safeBegin();
            }
        }
        [$ins, $er] = $this->flushBatch('facility_mappings_stage', $allCols, $buf);
        $valid += $ins; $errs += $er;
        $this->safeCommit();
        return [$total, $valid, $errs];
    }

    /** Load a Deductible visit-id file into deductible_mappings_stage (mirror of loadFacilityToStage). The
     *  file carries ONLY Visit Numbers; any other columns in the header are silently ignored. */
    private function loadDeductibleToStage(string $filePath, int $reportId): array
    {
        $map = [
            'visit number' => 'visit_id',
            'visit id'     => 'visit_id',
            'visitid'      => 'visit_id',
            'visit_id'     => 'visit_id',
        ];

        $iter = $this->iterateCsv($filePath);
        $iter->rewind();
        $header = $iter->current();
        if (!$header) throw new RuntimeException('Empty file.');

        $idxToCol = [];
        foreach ($header as $i => $h) {
            $k = strtolower(trim((string) $h));
            if (isset($map[$k])) $idxToCol[$i] = $map[$k];
        }
        if (!in_array('visit_id', $idxToCol, true)) throw new RuntimeException('Required column "Visit Number" not found in the deductible file.');

        $cols     = array_values(array_unique($idxToCol));   // = ['visit_id']
        $allCols  = array_merge(['report_id'], $cols);
        $visitPos = array_search('visit_id', $cols, true);
        $batchRows = max(1, min(5000, intdiv(120000, count($allCols))));

        $total = $valid = $errs = 0;
        $buf = []; $bufRows = 0;
        $this->safeBegin();
        $iter->next();
        while ($iter->valid()) {
            $row = $iter->current();
            $iter->next();
            $total++;
            if (!is_array($row)) continue;
            $values = array_fill(0, count($cols), null);
            foreach ($idxToCol as $i => $colName) {
                $colPos = array_search($colName, $cols, true);
                $values[$colPos] = $this->normalizeCellForFacility($colName, $row[$i] ?? null);   // trims the visit_id
            }
            if ($visitPos === false || $values[$visitPos] === null || $values[$visitPos] === '') { $errs++; continue; }
            $buf[] = $reportId;
            foreach ($values as $v) $buf[] = $v;
            if (++$bufRows >= $batchRows) {
                [$ins, $er] = $this->flushBatch('deductible_mappings_stage', $allCols, $buf);
                $valid += $ins; $errs += $er; $bufRows = 0;
                $this->safeCommit();
                $this->safeBegin();
            }
        }
        [$ins, $er] = $this->flushBatch('deductible_mappings_stage', $allCols, $buf);
        $valid += $ins; $errs += $er;
        $this->safeCommit();
        return [$total, $valid, $errs];
    }

    /* Defensive helpers — survive transient connection / batch issues */
    private function safeBegin(): void {
        if (!$this->pdo->inTransaction()) {
            try { $this->pdo->beginTransaction(); } catch (Throwable $t) {}
        }
    }
    private function safeCommit(): void {
        if ($this->pdo->inTransaction()) {
            try { $this->pdo->commit(); } catch (Throwable $t) {}
        }
    }

    /** True if a DB error is a dropped connection ("server has gone away" / "Lost connection"). */
    private function isGoneAway(Throwable $t): bool
    {
        $m = $t->getMessage();
        return stripos($m, 'gone away') !== false
            || stripos($m, 'Lost connection') !== false
            || strpos($m, '2006') !== false
            || strpos($m, '2013') !== false;
    }

    /** TRANSIENT lock contention — MySQL 1205 (lock wait timeout) / 1213 (deadlock). Both messages literally end
     *  with "try restarting transaction": they mean "someone else held the rows, do it again", NOT "this failed".
     *  They happen whenever two heavy writers touch the same report at once — e.g. a Clear running while an
     *  upload's post-append recompute is still going, or a user retrying Clear after the browser gave up while
     *  the first one was still running server-side (ignore_user_abort keeps it alive). Retrying the chunk is the
     *  correct response; aborting the whole operation, as we used to, is not. */
    private function isLockContention(Throwable $t): bool
    {
        $m = $t->getMessage();
        return strpos($m, '1205') !== false
            || strpos($m, '1213') !== false
            || stripos($m, 'Lock wait timeout') !== false
            || stripos($m, 'Deadlock found') !== false;
    }

    /** Re-establish a live PDO after a dropped connection and adopt it. */
    private function reconnect(): void
    {
        $this->pdo = db_reconnect();
    }

    /**
     * Insert a batch of rows as multi-row INSERTs.  $buf is a flat parameter array
     * of (rowCount × colCount) values.  The batch is split into sub-INSERTs each
     * kept well under max_allowed_packet so a large file with wide text rows can
     * never overflow the wire packet — the #1 cause of "MySQL server has gone away"
     * on big ingests when the server's global packet limit is small and we can't
     * raise it.  Each sub-INSERT also recovers from a dropped connection, then
     * falls back to per-row inserts so one bad row can't lose the chunk.
     * Returns [insertedRows, erroredRows] and clears $buf.
     */
    private function flushBatch(string $table, array $allCols, array &$buf): array
    {
        $colCount = count($allCols);
        $rowCount = $colCount > 0 ? intdiv(count($buf), $colCount) : 0;
        if ($rowCount === 0) { $buf = []; return [0, 0]; }

        $head   = "INSERT INTO $table (" . implode(',', $allCols) . ") VALUES ";
        $rowTpl = '(' . implode(',', array_fill(0, $colCount, '?')) . ')';

        $maxBytes = 1000000;            // ~1 MB of value bytes per INSERT (very conservative)
        $ins = $err = 0;
        $row = 0;
        while ($row < $rowCount) {
            $start = $row;
            $bytes = strlen($head) + 16;
            do {
                $base = $row * $colCount;
                $rb = $colCount * 3;     // per-row template/comma overhead
                for ($c = 0; $c < $colCount; $c++) {
                    $val = $buf[$base + $c];
                    $rb += ($val === null) ? 4 : strlen((string) $val) + 2;
                }
                $bytes += $rb;
                $row++;
            } while ($row < $rowCount && $bytes + 4096 < $maxBytes);

            $rowsHere = $row - $start;
            $slice    = array_slice($buf, $start * $colCount, $rowsHere * $colCount);
            [$i, $e]  = $this->insertChunk($head, $rowTpl, $colCount, $rowsHere, $slice);
            $ins += $i; $err += $e;
        }
        $buf = [];
        return [$ins, $err];
    }

    /** One multi-row INSERT with gone-away recovery, then per-row fallback. */
    private function insertChunk(string $head, string $rowTpl, int $colCount, int $rows, array $vals): array
    {
        $sql = $head . implode(',', array_fill(0, $rows, $rowTpl));
        try {
            $this->pdo->prepare($sql)->execute($vals);
            return [$rows, 0];
        } catch (Throwable $t) {
            if ($this->isGoneAway($t)) {
                // Connection dropped mid-stream — re-establish and retry the chunk once.
                try {
                    $this->reconnect();
                    $this->pdo->prepare($sql)->execute($vals);
                    return [$rows, 0];
                } catch (Throwable $t2) { /* fall through to per-row */ }
            }
            // Per-row fallback so a single bad/oversized row can't lose the chunk.
            $ins = $err = 0;
            for ($o = 0, $n = count($vals); $o < $n; $o += $colCount) {
                $one = array_slice($vals, $o, $colCount);
                try {
                    $this->pdo->prepare($head . $rowTpl)->execute($one);
                    $ins++;
                } catch (Throwable $e) {
                    if ($this->isGoneAway($e)) {
                        try { $this->reconnect(); $this->pdo->prepare($head . $rowTpl)->execute($one); $ins++; continue; }
                        catch (Throwable $e2) { /* counted as error below */ }
                    }
                    $err++;
                }
            }
            return [$ins, $err];
        }
    }

    /* ------------------------------------------------------------------ *
     * Derived columns
     * ------------------------------------------------------------------ */

    /**
     * After stage load, derive every computed column in as FEW full-table passes as possible —
     * 3 UPDATEs instead of the old 6 (+1 post-swap), which roughly halves the row-rewrite work
     * on a large file.  Output is byte-for-byte identical to the previous step-by-step version.
     */
    private function derivedCollectionOnStage(int $reportId): void
    {
        // Only ONE pass left — expected_payment.  dos_year/month, days, facility,
        // entry_date and upload_id are now pre-derived in PHP during loadCollectionToStage, so
        // those two full-table UPDATEs no longer run.  This pass still has to happen here because
        // the per-CPT rate needs the full file's aggregate — which only exists after load.
        // NOTE: `outstanding` is no longer derived; insurance_balance and patient_balance are
        // loaded directly from the CSV in loadCollectionToStage.
        $this->pdo->prepare("
            UPDATE collection_raw_stage cr
            LEFT JOIN (
                SELECT charge_code,
                       SUM(primary_ins_paid)
                         / NULLIF(SUM(CASE WHEN primary_ins_paid > 0 THEN COALESCE(units,1) ELSE 0 END), 0) AS rate
                FROM collection_raw_stage
                WHERE report_id = ? AND primary_ins_paid > 0
                  AND charge_code IS NOT NULL AND charge_code <> ''
                GROUP BY charge_code
            ) r ON r.charge_code = cr.charge_code
            SET cr.expected_payment = ROUND(COALESCE(r.rate, 0) * COALESCE(cr.units, 1), 2)
            WHERE cr.report_id = ?
        ")->execute([$reportId, $reportId]);
    }

    /**
     * Pre-compute every aggregate the Collection Dashboard needs and stash it
     * as JSON keyed by upload_id.  The dashboard reads from this cache (when
     * no filters are applied) instead of re-running aggregates on 200k+ rows.
     *
     * Cost: ~2 seconds run-once at ingest time.
     * Benefit: dashboard Overview tab loads in ~50 ms instead of ~1500 ms.
     */
    /*
     * Cross-report facility rollup: the physical mirror methods
     * (`mirrorFacilitiesForSource` and `refreshMirrorsForTarget`) used to copy rows
     * from each (source, facility) into the target report after every ingest.
     * That is gone — `collection_scope_for_report()` in admin/collection.php and
     * admin/collection-drill.php now expand the dashboard's WHERE clause at query
     * time to also match the mirror rules in `report_facility_mirror`.  No
     * duplicated rows in `collection_raw`, no DELETE/INSERT pass per upload,
     * source ingests are visible in rollup views immediately on the next page load.
     */

    public function computeAndCacheStats(int $uploadId, int $reportId): void
    {
        $reportId = (int) $reportId;   // hard-cast for safe interpolation
        $stats = [];

        /* All aggregates scope to this report via $reportId interpolation
           (the int is hard-cast above for safety).

           by_month and summary_subs share the SAME GROUP BY (dos_year,dos_month)
           and WHERE, so they are computed in ONE table scan and split in PHP.
           The former 'kpi' and 'carrier_mix' blocks were removed: the dashboard
           never read either key (it derives its totals from by_month), so they
           were pure cost. */
        $monthAgg = $this->pdo->query("
            SELECT dos_year, dos_month,
                   COUNT(DISTINCT visit_id) AS claims,
                   SUM(charge_amount) AS charge,
                   SUM(primary_ins_paid) AS pri,
                   SUM(secondary_ins_paid + tertiary_ins_paid) AS sec,
                   SUM(insurance_balance) AS bal,
                   SUM(patient_balance) AS pat_bal,
                   SUM(patient_payment_amount) AS pat_paid,
                   SUM(expected_payment) AS expected,
                   SUM(amount_allowed) AS allowed,
                   SUM(wo_amount) AS wo,
                   COUNT(DISTINCT CASE WHEN primary_ins_paid > 0 THEN visit_id END) AS paid_n,
                   (COUNT(DISTINCT visit_id) - COUNT(DISTINCT CASE WHEN primary_ins_paid > 0 THEN visit_id END)) AS unpaid_n,
                   COUNT(DISTINCT CASE WHEN primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) THEN visit_id END) AS prim_pend_n,
                   COUNT(DISTINCT CASE WHEN primary_ins_paid > 0 AND secondary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) THEN visit_id END) AS sec_pend_n,
                   COUNT(DISTINCT CASE WHEN (insurance_balance > 0 OR patient_balance > 0) THEN visit_id END) AS outst_n,
                   SUM(CASE WHEN primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) THEN insurance_balance + patient_balance ELSE 0 END) AS prim_outst,
                   SUM(CASE WHEN primary_ins_paid > 0 AND secondary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) THEN insurance_balance + patient_balance ELSE 0 END) AS sec_outst,
                   COUNT(DISTINCT CASE WHEN wo_amount > 0 AND primary_ins_paid = 0 AND secondary_ins_paid = 0 THEN visit_id END) AS u_c,
                   SUM(CASE WHEN wo_amount > 0 AND primary_ins_paid = 0 AND secondary_ins_paid = 0 THEN expected_payment ELSE 0 END) AS u_exp,
                   SUM(CASE WHEN wo_amount > 0 AND primary_ins_paid = 0 AND secondary_ins_paid = 0 THEN insurance_balance + patient_balance ELSE 0 END) AS u_out,
                   COUNT(DISTINCT CASE WHEN primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) AND charge_amount > 0 AND insurance_balance < charge_amount*0.4 THEN visit_id END) AS h_c,
                   SUM(CASE WHEN primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) AND charge_amount > 0 AND insurance_balance < charge_amount*0.4 THEN expected_payment ELSE 0 END) AS h_exp,
                   SUM(CASE WHEN primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) AND charge_amount > 0 AND insurance_balance < charge_amount*0.4 THEN insurance_balance + patient_balance ELSE 0 END) AS h_out,
                   COUNT(DISTINCT CASE WHEN primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) AND NOT (charge_amount > 0 AND insurance_balance < charge_amount*0.4) THEN visit_id END) AS un_c,
                   SUM(CASE WHEN primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) AND NOT (charge_amount > 0 AND insurance_balance < charge_amount*0.4) THEN charge_amount ELSE 0 END) AS un_charge,
                   SUM(CASE WHEN primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) AND NOT (charge_amount > 0 AND insurance_balance < charge_amount*0.4) THEN expected_payment ELSE 0 END) AS un_exp,
                   SUM(CASE WHEN primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) AND NOT (charge_amount > 0 AND insurance_balance < charge_amount*0.4) THEN insurance_balance + patient_balance ELSE 0 END) AS un_out
            FROM collection_raw
            WHERE report_id = $reportId AND dos_year IS NOT NULL AND dos_month IS NOT NULL
            GROUP BY dos_year, dos_month ORDER BY dos_year, dos_month
        ")->fetchAll();

        $byMonth = $summarySubs = [];
        foreach ($monthAgg as $r) {
            $byMonth[] = [
                'dos_year' => $r['dos_year'], 'dos_month' => $r['dos_month'],
                'claims' => $r['claims'], 'charge' => $r['charge'], 'pri' => $r['pri'], 'sec' => $r['sec'],
                'bal' => $r['bal'], 'pat_bal' => $r['pat_bal'], 'pat_paid' => $r['pat_paid'],
                'expected' => $r['expected'], 'allowed' => $r['allowed'], 'wo' => $r['wo'],
                'paid_n' => $r['paid_n'], 'unpaid_n' => $r['unpaid_n'],
                'prim_pend_n' => $r['prim_pend_n'], 'sec_pend_n' => $r['sec_pend_n'], 'outst_n' => $r['outst_n'],
                'prim_outst' => $r['prim_outst'], 'sec_outst' => $r['sec_outst'],
            ];
            $summarySubs[] = [
                'dos_year' => $r['dos_year'], 'dos_month' => $r['dos_month'],
                'u_c' => $r['u_c'], 'u_exp' => $r['u_exp'], 'u_out' => $r['u_out'],
                'h_c' => $r['h_c'], 'h_exp' => $r['h_exp'], 'h_out' => $r['h_out'],
                'un_c' => $r['un_c'], 'un_charge' => $r['un_charge'], 'un_exp' => $r['un_exp'], 'un_out' => $r['un_out'],
            ];
        }
        $stats['by_month'] = $byMonth;
        $stats['summary_subs'] = $summarySubs;

        /* top_facilities is derived from uniq_pt_facility below (same GROUP BY
           facility) instead of its own scan + a second per-facility uniq query.
           The dead carrier_mix block (never read by the dashboard) was removed. */

        $stats['facets'] = [
            'facilities' => $this->pdo->query("SELECT DISTINCT facility FROM collection_raw WHERE report_id = $reportId AND facility IS NOT NULL AND facility <> '' ORDER BY facility")->fetchAll(PDO::FETCH_COLUMN),
            'providers'  => $this->pdo->query("SELECT DISTINCT provider_name FROM collection_raw WHERE report_id = $reportId AND provider_name IS NOT NULL AND provider_name <> '' ORDER BY provider_name")->fetchAll(PDO::FETCH_COLUMN),
            'years'      => array_map('intval', $this->pdo->query("SELECT DISTINCT dos_year FROM collection_raw WHERE report_id = $reportId AND dos_year IS NOT NULL ORDER BY dos_year DESC")->fetchAll(PDO::FETCH_COLUMN)),
        ];

        /* One scan over GROUP BY facility feeds BOTH uniq_pt_facility and
           top_facilities.  pri/sec are pulled here so top_facilities can be
           derived without its own scan; they are stripped from the cached
           uniq_pt_facility rows so that key stays byte-for-byte identical. */
        $facBase = $this->pdo->query("
            SELECT COALESCE(facility,'(unmapped)') AS facility,
                   COUNT(DISTINCT chart_number) AS uniq_pt,
                   COUNT(DISTINCT provider_name) AS providers,
                   COUNT(DISTINCT visit_id) AS visits, SUM(charge_amount) AS charge,
                   SUM(primary_ins_paid + secondary_ins_paid + tertiary_ins_paid) AS paid,
                   SUM(insurance_balance) AS bal, SUM(patient_balance) AS pat_bal,
                   SUM(expected_payment) AS expected,
                   SUM(primary_ins_paid) AS pri,
                   SUM(secondary_ins_paid + tertiary_ins_paid) AS sec
            FROM collection_raw WHERE report_id = $reportId
            GROUP BY facility ORDER BY uniq_pt DESC
        ")->fetchAll();

        $stats['uniq_pt_facility'] = array_map(fn($r) => [
            'facility' => $r['facility'], 'uniq_pt' => $r['uniq_pt'], 'providers' => $r['providers'],
            'visits' => $r['visits'], 'charge' => $r['charge'], 'paid' => $r['paid'],
            'bal' => $r['bal'], 'pat_bal' => $r['pat_bal'], 'expected' => $r['expected'],
        ], $facBase);

        /* top_facilities = top 10 facilities by charge.  Mirrors the original
           query exactly: claims = COUNT(DISTINCT visit_id) (= visits), and uniq_pt
           is 0 for the '(unmapped)' bucket (the original second-pass query skipped
           it, so it defaulted to 0). */
        $facByCharge = $facBase;
        usort($facByCharge, fn($a, $b) => (float)$b['charge'] <=> (float)$a['charge']);
        $stats['top_facilities'] = array_map(fn($r) => [
            'facility' => $r['facility'], 'claims' => $r['visits'], 'charge' => $r['charge'],
            'pri' => $r['pri'], 'sec' => $r['sec'], 'bal' => $r['bal'], 'pat_bal' => $r['pat_bal'],
            'expected' => $r['expected'],
            'uniq_pt' => $r['facility'] === '(unmapped)' ? 0 : $r['uniq_pt'],
        ], array_slice($facByCharge, 0, 10));

        $stats['uniq_pt_facility_provider'] = $this->pdo->query("
            SELECT COALESCE(facility,'(unmapped)') AS facility,
                   COALESCE(provider_name,'(no provider)') AS provider,
                   COUNT(DISTINCT chart_number) AS uniq_pt,
                   COUNT(DISTINCT visit_id) AS visits, SUM(charge_amount) AS charge,
                   SUM(primary_ins_paid + secondary_ins_paid + tertiary_ins_paid) AS paid,
                   SUM(insurance_balance) AS bal,
                   SUM(patient_balance) AS pat_bal
            FROM collection_raw WHERE report_id = $reportId
            GROUP BY facility, provider_name ORDER BY facility, uniq_pt DESC
        ")->fetchAll();

        $stats['uniq_pt_facility_provider_month'] = $this->pdo->query("
            SELECT COALESCE(facility,'(unmapped)') AS facility,
                   COALESCE(provider_name,'(no provider)') AS provider,
                   dos_year, dos_month,
                   COUNT(DISTINCT chart_number) AS uniq_pt,
                   COUNT(DISTINCT visit_id) AS visits, SUM(charge_amount) AS charge,
                   SUM(primary_ins_paid + secondary_ins_paid + tertiary_ins_paid) AS paid,
                   SUM(insurance_balance) AS bal,
                   SUM(patient_balance) AS pat_bal
            FROM collection_raw
            WHERE report_id = $reportId AND dos_year IS NOT NULL AND dos_month IS NOT NULL
            GROUP BY facility, provider_name, dos_year, dos_month
            ORDER BY facility, provider_name, dos_year, dos_month
        ")->fetchAll();

        /* summary_subs is produced together with by_month in the single merged
           scan near the top of this method. */

        $bucketCase = "CASE
            WHEN DATEDIFF(CURDATE(), dos) <= 30  THEN '0-30'
            WHEN DATEDIFF(CURDATE(), dos) <= 60  THEN '31-60'
            WHEN DATEDIFF(CURDATE(), dos) <= 90  THEN '61-90'
            WHEN DATEDIFF(CURDATE(), dos) <= 120 THEN '91-120'
            ELSE '121+' END";

        $stats['aging_buckets'] = $this->pdo->query("
            SELECT bucket, COUNT(DISTINCT visit_id) AS claims,
                   SUM(ins_bal) AS bal, SUM(pat_bal) AS pat_bal,
                   SUM(charge_amount) AS charge, SUM(amount_allowed) AS expected
            FROM (SELECT visit_id, $bucketCase AS bucket,
                         insurance_balance AS ins_bal, patient_balance AS pat_bal,
                         charge_amount, amount_allowed
                  FROM collection_raw
                  WHERE report_id = $reportId
                    AND (insurance_balance > 0 OR patient_balance > 0)
                    AND dos IS NOT NULL) t
            GROUP BY bucket ORDER BY FIELD(bucket,'0-30','31-60','61-90','91-120','121+')
        ")->fetchAll();

        $stats['aging_by_carrier'] = $this->pdo->query("
            SELECT COALESCE(carrier1_name,'(none)') AS carrier, bucket,
                   COUNT(DISTINCT visit_id) AS claims,
                   SUM(ins_bal) AS bal, SUM(pat_bal) AS pat_bal
            FROM (SELECT visit_id, carrier1_name, $bucketCase AS bucket,
                         insurance_balance AS ins_bal, patient_balance AS pat_bal
                  FROM collection_raw
                  WHERE report_id = $reportId
                    AND (insurance_balance > 0 OR patient_balance > 0)
                    AND dos IS NOT NULL) t
            GROUP BY carrier, bucket
        ")->fetchAll();

        $stats['aging_by_facility'] = $this->pdo->query("
            SELECT COALESCE(facility,'(unmapped)') AS facility, bucket,
                   COUNT(DISTINCT visit_id) AS claims,
                   SUM(ins_bal) AS bal, SUM(pat_bal) AS pat_bal
            FROM (SELECT visit_id, facility, $bucketCase AS bucket,
                         insurance_balance AS ins_bal, patient_balance AS pat_bal
                  FROM collection_raw
                  WHERE report_id = $reportId
                    AND (insurance_balance > 0 OR patient_balance > 0)
                    AND dos IS NOT NULL) t
            GROUP BY facility, bucket
        ")->fetchAll();

        $stats['carrier_performance'] = $this->pdo->query("
            SELECT COALESCE(carrier1_name,'(none)') AS carrier,
                   COUNT(DISTINCT visit_id) AS claims, COUNT(DISTINCT chart_number) AS patients,
                   SUM(charge_amount) AS charge, SUM(amount_allowed) AS allowed,
                   SUM(primary_ins_paid) AS paid,
                   SUM(insurance_balance) AS bal, SUM(patient_balance) AS pat_bal,
                   SUM(primary_ins_paid + secondary_ins_paid + tertiary_ins_paid) AS ins_paid,
                   SUM(expected_payment) AS expected,
                   COUNT(DISTINCT CASE WHEN primary_ins_paid > 0 THEN visit_id END) AS paid_n,
                   COUNT(DISTINCT CASE WHEN primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) THEN visit_id END) AS unpaid_n,
                   AVG(days_to_first_payment) AS avg_days
            FROM collection_raw WHERE report_id = $reportId
            GROUP BY carrier ORDER BY charge DESC
        ")->fetchAll();

        $cptRows = $this->pdo->query("
            SELECT cr.charge_code, MAX(cr.charge_code_desc) AS description,
                   COUNT(*) AS volume,
                   COUNT(DISTINCT cr.chart_number) AS patients,
                   SUM(cr.units) AS units, SUM(cr.charge_amount) AS charge,
                   SUM(cr.amount_allowed) AS allowed,
                   SUM(cr.primary_ins_paid + cr.secondary_ins_paid + cr.tertiary_ins_paid) AS paid,
                   SUM(cr.expected_payment) AS expected,
                   SUM(cr.insurance_balance) AS bal,
                   SUM(cr.patient_balance) AS pat_bal,
                   SUM(CASE WHEN cr.primary_ins_paid = 0 THEN 1 ELSE 0 END) AS unpaid_n
            FROM collection_raw cr
            WHERE cr.report_id = $reportId AND cr.charge_code IS NOT NULL
              AND cr.charge_code <> '' AND cr.charge_code NOT LIKE '#%'
            GROUP BY cr.charge_code ORDER BY charge DESC
        ")->fetchAll();
        $stats['cpt_analysis'] = $cptRows;

        $stats['provider_performance'] = $this->pdo->query("
            SELECT provider_name, COUNT(DISTINCT visit_id) AS claims,
                   COUNT(DISTINCT chart_number) AS patients,
                   COUNT(DISTINCT facility) AS facilities,
                   COUNT(DISTINCT charge_code) AS cpts,
                   SUM(charge_amount) AS charge, SUM(amount_allowed) AS allowed,
                   SUM(primary_ins_paid + secondary_ins_paid + tertiary_ins_paid) AS paid,
                   SUM(expected_payment) AS expected,
                   SUM(insurance_balance) AS bal, SUM(patient_balance) AS pat_bal,
                   AVG(days_to_first_payment) AS avg_days,
                   COUNT(DISTINCT CASE WHEN primary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) THEN visit_id END) AS prim_pend_n,
                   COUNT(DISTINCT CASE WHEN primary_ins_paid > 0 AND secondary_ins_paid = 0 AND (insurance_balance > 0 OR patient_balance > 0) THEN visit_id END) AS sec_pend_n
            FROM collection_raw
            WHERE report_id = $reportId AND provider_name IS NOT NULL AND provider_name <> ''
            GROUP BY provider_name ORDER BY charge DESC
        ")->fetchAll();

        /* ============= PIVOT (Facility → Month → Day) — pre-aggregated ============= */
        /* `b` is total balance (insurance + patient) — single combined column for the pivot. */
        $stats['pivot_fmd'] = $this->pdo->query("
            SELECT COALESCE(facility,'(unmapped)') AS facility,
                   dos_year, dos_month, DAY(dos) AS dos_day,
                   COUNT(*) AS v, SUM(charge_amount) AS c,
                   SUM(primary_ins_paid) AS p,
                   SUM(secondary_ins_paid + tertiary_ins_paid) AS s,
                   SUM(insurance_balance + patient_balance) AS b
            FROM collection_raw
            WHERE report_id = $reportId AND dos IS NOT NULL
            GROUP BY facility, dos_year, dos_month, dos_day
            ORDER BY facility, dos_year, dos_month, dos_day
        ")->fetchAll();

        /* ============= GRAND TOTAL (used by Unique Patients tab) ============= */
        $stats['unique_grand_total'] = $this->pdo->query("
            SELECT COUNT(DISTINCT chart_number) AS uniq_pt,
                   COUNT(DISTINCT visit_id) AS visits,
                   SUM(charge_amount) AS charge,
                   SUM(primary_ins_paid + secondary_ins_paid + tertiary_ins_paid) AS paid,
                   SUM(insurance_balance) AS bal,
                   SUM(patient_balance) AS pat_bal
            FROM collection_raw WHERE report_id = $reportId
        ")->fetch();

        /* ============= RECENTLY BILLED (by Entry Date — last 30 days of the newest entry) ============= */
        /* `outstanding` here is the total balance (ins + pat) — the cache shape keeps the
            'outstanding' key name for back-compat with the dashboard tile that reads it. */
        $latestEntry = $this->pdo->query("SELECT MAX(entry_date) FROM collection_raw WHERE report_id = $reportId")->fetchColumn();
        $rb = ['claims' => 0, 'outstanding' => 0.0, 'charge' => 0.0, 'latest_entry' => $latestEntry ?: null, 'since' => null];
        if ($latestEntry) {
            $st = $this->pdo->prepare("
                SELECT COUNT(DISTINCT visit_id) AS claims,
                       SUM(insurance_balance + patient_balance) AS outstanding,
                       SUM(charge_amount) AS charge
                FROM collection_raw
                WHERE report_id = ? AND entry_date IS NOT NULL AND entry_date >= DATE_SUB(?, INTERVAL 30 DAY)");
            $st->execute([$reportId, $latestEntry]);
            $r = $st->fetch();
            $rb['claims'] = (int) $r['claims'];
            $rb['outstanding'] = (float) $r['outstanding'];
            $rb['charge'] = (float) $r['charge'];
            $rb['since'] = date('Y-m-d', strtotime($latestEntry . ' -30 days'));
        }
        $stats['recently_billed'] = $rb;

        /* Store as single JSON blob (REPLACE = upsert).  Index by upload_id but tag report_id for safety.
           Recover once if the (potentially large) blob trips a dropped connection. */
        $blob = json_encode($stats);
        // Guard: never SAVE an all-empty snapshot when the report actually has rows. An empty snapshot is a failed /
        // partial compute (e.g. during a temp-disk incident); saving it blanks the filter facets + Compare tab. Skipping
        // it leaves the report's snapshot absent, so admin/collection.php falls back to live SQL (correct, just slower).
        // A genuinely EMPTY report (0 rows) is allowed to save an empty snapshot (it's accurate). $reportId is int-cast.
        if (empty($stats['by_month']) && empty($stats['facets']['facilities']) && empty($stats['facets']['providers'])) {
            $hasRows = 0;
            try { $hasRows = (int) $this->pdo->query("SELECT EXISTS(SELECT 1 FROM collection_raw WHERE report_id = $reportId)")->fetchColumn(); } catch (Throwable $e) { $hasRows = 0; }
            if ($hasRows) {
                try { log_error_db('computeAndCacheStats produced empty stats for a non-empty report — snapshot save skipped (live fallback kept).', 'ingest', ['report_id' => $reportId, 'upload_id' => $uploadId], 'warning'); } catch (Throwable $e) {}
                return;   // do not overwrite with a bogus empty snapshot
            }
        }
        $sql  = "REPLACE INTO collection_stats (upload_id, snapshot, computed_at) VALUES (?, ?, NOW())";
        try {
            $this->pdo->prepare($sql)->execute([$uploadId, $blob]);
        } catch (Throwable $t) {
            if ($this->isGoneAway($t)) { $this->reconnect(); $this->pdo->prepare($sql)->execute([$uploadId, $blob]); }
            else { throw $t; }
        }

        // Housekeeping: keep only the most recent few snapshots for this report. Every upload writes a new
        // collection_stats row (keyed by upload_id); once superseded it is never read again, so without this they
        // accumulate forever. The just-saved snapshot has the highest upload_id, so it is always among those kept.
        // Best-effort — a prune failure must NEVER fail the upload. $reportId is int-cast at the top of this method,
        // so it interpolates safely (no bound params → no native-prepare placeholder issues).
        try {
            $keep = 3;
            $this->pdo->exec("DELETE cs FROM collection_stats cs
                JOIN collection_uploads cu ON cu.id = cs.upload_id
                WHERE cu.report_id = $reportId
                  AND cs.upload_id NOT IN (
                      SELECT id FROM (
                          SELECT cu2.id FROM collection_uploads cu2
                          JOIN collection_stats cs2 ON cs2.upload_id = cu2.id
                          WHERE cu2.report_id = $reportId
                          ORDER BY cu2.id DESC LIMIT $keep
                      ) k
                  )");
        } catch (Throwable $e) { /* best-effort housekeeping — never fail the upload over cleanup */ }
    }

    /** Re-apply THIS report's facility mapping to its collection rows. */
    public function applyFacilityToCollection(int $reportId): void
    {
        $this->pdo->prepare("
            UPDATE collection_raw cr
            JOIN (
                SELECT visit_id, MAX(facility) AS facility, MAX(entry_date) AS entry_date
                FROM facility_mappings
                WHERE report_id = ? AND facility IS NOT NULL AND facility <> ''
                GROUP BY visit_id
            ) fm ON fm.visit_id = cr.visit_id
            SET cr.facility = fm.facility, cr.entry_date = fm.entry_date
            WHERE cr.report_id = ?
        ")->execute([$reportId, $reportId]);
    }

    /** Re-apply THIS report's deductible visit list to its collection rows. FULL REPLACE: reset the
     *  currently-flagged rows to 0, then flag every visit in the uploaded list — so a re-upload that drops
     *  a visit un-flags it (the uploaded list is the complete set of deductible visits). Two cheap UPDATEs;
     *  the reset only touches already-flagged rows so it never rewrites the whole report. */
    public function applyDeductibleToCollection(int $reportId): void
    {
        $this->pdo->prepare("UPDATE collection_raw SET deductible = 0 WHERE report_id = ? AND deductible <> 0")
                  ->execute([$reportId]);
        $this->pdo->prepare("
            UPDATE collection_raw cr
            JOIN (
                SELECT DISTINCT visit_id FROM deductible_mappings WHERE report_id = ?
            ) dm ON dm.visit_id = cr.visit_id
            SET cr.deductible = 1
            WHERE cr.report_id = ?
        ")->execute([$reportId, $reportId]);
    }


    /* ------------------------------------------------------------------ *
     * Cell normalizers
     * ------------------------------------------------------------------ */
    private function normalizeCellForCollection(string $col, $raw)
    {
        $v = is_string($raw) ? trim($raw) : $raw;

        // Carriers are ACCEPTED even when empty/unknown and canonicalised to "N/A"
        // (never stored blank) so the dashboards group/label them consistently.
        // The ops team no longer has to type "N/A" by hand — the system does it.
        if (in_array($col, ['carrier1_code','carrier1_name','carrier2_code','carrier2_name','carrier3_code','carrier3_name'], true)) {
            $low = strtolower(trim((string) ($v ?? '')));
            if ($low === '' || in_array($low, ['n/a', 'na', 'none', 'unknown', 'null', '-', '--'], true)) return 'N/A';
            return trim((string) $v);
        }

        if ($v === '' || $v === null) return null;
        switch ($col) {
            case 'dos':
            case 'primary_ins_payment_date':
            case 'secondary_ins_payment_date':
            case 'tertiary_ins_payment_date':
            case 'patient_payment_date':
            case 'wo_date':
            case 'appointment_date':
                return $this->parseDate((string)$v);
            case 'units':
            case 'charge_amount':
            case 'amount_allowed':
            case 'primary_ins_paid':
            case 'secondary_ins_paid':
            case 'tertiary_ins_paid':
            case 'patient_payment_amount':
            case 'wo_amount':
            case 'patient_balance':
            case 'insurance_balance':
                return $this->parseNumber((string)$v);
            case 'visit_id':
            case 'chart_number':
                return (string) trim((string)$v);
            case 'charge_code':
                return strtoupper(trim((string)$v));
            default:
                return is_string($v) ? trim($v) : $v;
        }
    }

    private function normalizeCellForFacility(string $col, $raw)
    {
        $v = is_string($raw) ? trim($raw) : $raw;
        if ($v === '' || $v === null) return null;
        if ($col === 'visit_id') return (string) trim((string)$v);
        if ($col === 'entry_date') return $this->parseDate((string)$v);
        return is_string($v) ? trim($v) : $v;
    }

    private function parseDate(string $v): ?string
    {
        $v = trim($v);
        if ($v === '') return null;
        // Excel serial fallback
        if (is_numeric($v) && $v > 1 && $v < 100000) {
            $unix = -2208988800 + ($v - 1) * 86400;
            if ($v >= 60) $unix -= 86400;
            return gmdate('Y-m-d', (int) round($unix));
        }
        foreach (['Y-m-d', 'Y-m-d H:i:s', 'm/d/Y', 'n/j/Y', 'd/m/Y', 'd-m-Y', 'd-M-Y', 'M j, Y'] as $fmt) {
            $dt = DateTime::createFromFormat($fmt, $v);
            if ($dt !== false) return $dt->format('Y-m-d');
        }
        $ts = strtotime($v);
        return $ts ? date('Y-m-d', $ts) : null;
    }

    private function parseNumber(string $v): float
    {
        return (float) preg_replace('/[^0-9.\-]/', '', $v);
    }

    /* ------------------------------------------------------------------ *
     * Upload records
     * ------------------------------------------------------------------ */
    private function createUploadRecord(string $kind, string $filePath, int $userId, ?int $year = null, ?int $reportId = null): int
    {
        $hash = @hash_file('sha256', $filePath) ?: null;
        $stmt = $this->pdo->prepare(
            "INSERT INTO collection_uploads
              (report_id, kind, file_name, file_path, file_size, file_hash, year, status, uploaded_by, started_at, created_at, updated_at)
             VALUES (?,?,?,?,?,?,?,'processing',?,NOW(),NOW(),NOW())"
        );
        $stmt->execute([
            $reportId,
            $kind,
            basename($filePath),
            $filePath,
            @filesize($filePath) ?: null,
            $hash,
            $year,
            $userId,
        ]);
        return (int) $this->pdo->lastInsertId();
    }

    private function completeUploadRecord(int $id, int $total, int $valid, int $errs): void
    {
        $this->pdo->prepare(
            'UPDATE collection_uploads
                SET total_rows=?, valid_rows=?, error_rows=?, status="completed",
                    completed_at=NOW(), updated_at=NOW()
              WHERE id=?'
        )->execute([$total, $valid, $errs, $id]);
    }

    private function failUploadRecord(int $id, string $msg): void
    {
        $this->pdo->prepare(
            'UPDATE collection_uploads
                SET status="failed", note=?, updated_at=NOW()
              WHERE id=?'
        )->execute([mb_substr($msg, 0, 500), $id]);
    }
}
