From ed19f51a52514cbd9790fd7ffe4ef6204fccb03a Mon Sep 17 00:00:00 2001
From: Ricardo Mancinas <rmancinas@freakma.net>
Date: Tue, 4 Aug 2026 22:20:30 -0700
Subject: [PATCH] fix(ops): re-stage before an additive sync
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

SYNC ran `run_all.py --sync` without `--stage`, so it depended on staged
Parquet under migration/output. That directory is part of the image, not a
volume, so any redeploy wiped it and the job died on the first transform:

    FileNotFoundError: '/repo/migration/output/stg_utilities/datgral.parquet'

Re-staging is also what makes the job's own label true — without it a sync
would replay whatever upload staged last, not the files currently in the
ingest folder.

Staging now counts as a numbered step when it runs, so the Operaciones
progress bar moves during the slowest phase instead of sitting empty.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 apps/api/src/ops/ops.service.ts |  7 ++++++-
 migration/run_all.py            | 22 +++++++++++++++++-----
 2 files changed, 23 insertions(+), 6 deletions(-)

diff --git a/apps/api/src/ops/ops.service.ts b/apps/api/src/ops/ops.service.ts
index 3d754cf..8b53cc9 100644
--- a/apps/api/src/ops/ops.service.ts
+++ b/apps/api/src/ops/ops.service.ts
@@ -400,7 +400,12 @@ export class OpsService implements OnModuleInit {
         `${PIPEFAIL}echo '== Respaldo de seguridad previo ==' && ` +
         `${this.dumpCommand(flags, db, out)} && ` +
         `echo '== Sincronización aditiva desde carpeta de ingesta ==' && ` +
-        `${shq(py)} ${runAll} --env ${shq(this.migrationEnv)} --sync`;
+        // --stage is not optional here. The staged Parquet lives in the image
+        // at migration/output, NOT on a volume, so every redeploy wipes it and
+        // a sync without --stage dies on a missing stg_*/*.parquet. Re-staging
+        // is also the only thing that makes "desde carpeta de ingesta" true:
+        // stale Parquet would sync the previous upload, not the current one.
+        `${shq(py)} ${runAll} --env ${shq(this.migrationEnv)} --stage --sync`;
       return { cmd, resolvedParams: { safetyBackup: file } };
     }
 
diff --git a/migration/run_all.py b/migration/run_all.py
index 1e68f34..c53c70c 100644
--- a/migration/run_all.py
+++ b/migration/run_all.py
@@ -15,6 +15,11 @@ Then:
   ./.venv/bin/python run_all.py --env dev            # data only (staging already present)
   ./.venv/bin/python run_all.py --env prod --stage   # re-extract from Access first, then load
 
+--sync swaps the truncate+rebuild steps for the additive upsert ones. It reads
+the same staged Parquet, so it needs --stage too unless a previous run left
+migration/output populated on this machine — which is never true in a
+container, where that directory is part of the image and dies with it.
+
 Reproducing dev -> prod is exactly `--env prod` (plus --stage if the staged
 Parquet isn't present on the machine running it).
 
@@ -100,15 +105,22 @@ def main() -> None:
                     help="upsert legacy rows and archive removed legacy rows; preserve manual rows")
     args = ap.parse_args()
 
-    if args.stage:
-        run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")])
-
     steps = SYNC_STEPS if args.sync else STEPS
-    for i, step in enumerate(steps, start=1):
+    # Staging counts as a step when it runs: it is the slowest part of the pass
+    # (mdbtools re-reads every Access file), so leaving it outside the numbering
+    # would park the Operaciones progress bar at "nothing yet" for minutes.
+    total = len(steps) + (1 if args.stage else 0)
+    offset = 1 if args.stage else 0
+
+    if args.stage:
+        run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")],
+            step=1, total=total)
+
+    for i, step in enumerate(steps, start=1 + offset):
         cmd = [PY, str(HERE / step), "--env", args.env]
         if args.sync:
             cmd.append("--sync")
-        run(cmd, step=i, total=len(steps))
+        run(cmd, step=i, total=total)
 
     print(f"\n✓ migration complete for env={args.env}")