chore(release): v1.0.15

Cut by rmancinas via the "Cut release" workflow. Pushing the tag triggers build.yml; deploy separately with tag=1.0.15.

.gitea/workflows/build.yml

@@ -14,7 +14,16 @@
 # ARG/ENV (APP_VERSION / GIT_SHA / BUILD_DATE) and as OCI labels, so a running
 # container can report exactly what is deployed.
 #
-# Release flow:  git tag v1.2.0 && git push origin v1.2.0  -> versioned images.
+# Release flow:  git tag v1.2.0 && git push origin v1.2.0  -> versioned images
+#                -> deploy-on-tag.yml waits for this run to go green and then
+#                   dispatches deploy-galactus.yml.
+#
+# This workflow BUILDS ONLY — it never deploys. The deploy chain used to be a
+# job here, gated to tag refs, but Gitea draws every job of a workflow into the
+# run graph before it evaluates the job's `if`: a routine master build showed a
+# pending "Deploy to galactus" and looked like prod was about to be redeployed
+# off an unreleased commit. Keeping the deploy in a `on: push: tags` workflow of
+# its own makes that structurally impossible.
 
 name: Build and Push Images
 

.gitea/workflows/deploy-galactus.yml

@@ -295,6 +295,9 @@ jobs:
               "SESSION_COOKIE_SECURE": "false",
               "OPS_DB_ADMIN_USER": "root",
               "OPS_DB_ADMIN_PASSWORD": "${{ secrets.MYSQL_ROOT_PASSWORD }}",
+              "REPLICA_DB_HOST": "${{ secrets.REPLICA_DB_HOST }}",
+              "REPLICA_DB_USER": "${{ secrets.REPLICA_DB_USER }}",
+              "REPLICA_DB_PASS": "${{ secrets.REPLICA_DB_PASS }}",
               "MINIO_ROOT_USER": "${{ secrets.MINIO_ROOT_USER }}",
               "MINIO_ROOT_PASSWORD": "${{ secrets.MINIO_ROOT_PASSWORD }}",
               "SES_REGION": "${{ secrets.SES_REGION }}",

.gitea/workflows/deploy-on-tag.yml

@@ -0,0 +1,199 @@
+# Chain the PROD deploy onto a green tag build.
+#
+# This is a SEPARATE workflow, not a job in build.yml, and the trigger is the
+# whole point: `on: push: tags` cannot fire on a push to master. When this was a
+# `deploy` job inside build.yml gated by `if: startsWith(github.ref,
+# 'refs/tags/v')`, Gitea still drew "Deploy to galactus" into the job graph of
+# every ordinary master build — the `if` is not evaluated until `needs` resolve,
+# so the job sits there looking like an imminent production deploy on a commit
+# nobody released. That is indistinguishable from a real misfire, and the only
+# safe reaction is to cancel the run, which kills the images with it.
+#
+# What it does NOT do is build. build.yml already builds and pushes both images
+# from one run; this waits for that run to go green and then dispatches
+# deploy-galactus.yml, which only pulls.
+#
+# Why wait for the build run rather than just dispatching: deploy-galactus.yml
+# pulls api and web at the same tag, and a half-pushed pair is exactly the state
+# that leaves prod running one new image and one old one. The build run turning
+# green is the signal that both are in the registry.
+#
+# Why a dispatch and not a `workflow_run:` trigger, which Gitea does support as
+# of 1.24: deploy-galactus.yml reads `github.event.inputs.*` in ten places (tag,
+# scope, bootstrap, skip_migrate). Under workflow_run every one of them is the
+# empty string, so the deploy would silently run with no tag and scope != 'full'.
+# A dispatch keeps that workflow's contract intact and keeps it hand-runnable for
+# rollbacks, which is the whole point of it.
+#
+# Kill switch: set the repo variable AUTO_DEPLOY_GALACTUS to `false` to cut the
+# chain and go back to dispatching the deploy by hand. Anything else (including
+# unset) deploys.
+
+name: Deploy on tag
+
+on:
+  push:
+    tags: ["v*"]
+
+jobs:
+  deploy:
+    name: Deploy to galactus
+    runs-on: docker
+    container:
+      image: node:20-alpine
+    steps:
+      - name: Preflight — RELEASE_TOKEN
+        env:
+          RELEASE_TOKEN: ${{ secrets.RELEASE_TOKEN }}
+        run: |
+          set -eu
+          if [ -z "${RELEASE_TOKEN:-}" ]; then
+            echo "::error::Secret RELEASE_TOKEN is not set, so this cannot wait"
+            echo "::error::for the build or dispatch the deploy. Once build.yml"
+            V=${GITHUB_REF#refs/tags/}
+            echo "::error::is green, run 'Deploy to galactus' by hand with tag=${V#v}."
+            exit 1
+          fi
+
+      - name: Wait for the tag build, then dispatch deploy-galactus.yml
+        env:
+          RELEASE_TOKEN: ${{ secrets.RELEASE_TOKEN }}
+          AUTO_DEPLOY: ${{ vars.AUTO_DEPLOY_GALACTUS }}
+          TAG_REF: ${{ github.ref }}
+          BUILD_SHA: ${{ github.sha }}
+        run: |
+          node -e '
+            const base = `${process.env.GITHUB_SERVER_URL}/api/v1/repos/${process.env.GITHUB_REPOSITORY}`;
+            const headers = { Authorization: `token ${process.env.RELEASE_TOKEN}` };
+            // refs/tags/v1.2.3 — derived from github.ref rather than ref_name so
+            // it does not depend on how Gitea populates GITHUB_REF_NAME.
+            const tagRef = process.env.TAG_REF;
+            const tag = tagRef.replace(/^refs\/tags\//, "");
+            // The git tag carries the leading v; the image tag does not.
+            const version = tag.replace(/^v/, "");
+            const sha = process.env.BUILD_SHA;
+            const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+
+            const runs = async () => {
+              const r = await fetch(`${base}/actions/runs?limit=50`, { headers });
+              if (!r.ok) throw new Error(`runs query failed: HTTP ${r.status}`);
+              return (await r.json()).workflow_runs || [];
+            };
+
+            // The release commit and its tag are the SAME sha, and build.yml
+            // skips the master run by design — so a sha match alone can latch
+            // onto that skipped run and call the build green when no image was
+            // ever pushed. Require the tag ref when the API reports one.
+            const isTagRun = (r) => {
+              const ref = r.head_branch || r.ref || "";
+              return !ref || ref === tag || ref === tagRef;
+            };
+
+            const buildRun = async () =>
+              (await runs()).find(
+                (r) =>
+                  r.head_sha === sha &&
+                  String(r.path || "").includes("build.yml") &&
+                  isTagRun(r),
+              );
+
+            // Gitea reports a run as `status` and mirrors it into `conclusion`;
+            // read whichever is populated rather than betting on one field.
+            const outcome = (r) => String(r.conclusion || r.status || "").toLowerCase();
+            const DONE = ["success", "failure", "cancelled", "canceled", "skipped"];
+
+            // Every deploy-galactus run id visible right now. A dispatch is only
+            // confirmed by an id that is NOT in here — a plain "is there a deploy
+            // run" check is satisfied by the PREVIOUS release run, and would
+            // report success for a dispatch that never took.
+            const deployRunIds = async () =>
+              new Set(
+                (await runs())
+                  .filter((r) => String(r.path || "").includes("deploy-galactus.yml"))
+                  .map((r) => r.id),
+              );
+
+            (async () => {
+              if (process.env.AUTO_DEPLOY === "false") {
+                console.log("AUTO_DEPLOY_GALACTUS=false — not deploying.");
+                console.log(`Deploy by hand with tag=${version} when ready.`);
+                return;
+              }
+
+              // ~20 min. A build is about 90s; the rest is queue time behind
+              // other runs on a single runner.
+              let run = null;
+              for (let i = 0; i < 80; i++) {
+                run = await buildRun();
+                if (run && DONE.includes(outcome(run))) break;
+                if (!run && i === 11) {
+                  // Two minutes with no run at all. The post-receive hook drops
+                  // runs silently when it errors (this cost v1.0.3 its images),
+                  // so say so rather than timing out with no explanation.
+                  console.log(`::warning::No build.yml run for ${tag} yet after 2 min.`);
+                  console.log(`::warning::If the Gitea post-receive hook is broken, dispatch`);
+                  console.log(`::warning::"Build and Push Images" by hand with ref=${tag}.`);
+                }
+                await sleep(15_000);
+              }
+
+              if (!run) {
+                console.log(`::error::No build.yml run for ${tag} (${sha}) after 20 min.`);
+                console.log(`::error::Dispatch "Build and Push Images" with ref=${tag} (the`);
+                console.log(`::error::tag, not master), then deploy by hand with tag=${version}.`);
+                process.exit(1);
+              }
+
+              const result = outcome(run);
+              if (result !== "success") {
+                console.log(`::error::build.yml for ${tag} ended as "${result}" — not deploying.`);
+                console.log(`::error::Fix the build, re-run it, then deploy by hand with tag=${version}.`);
+                process.exit(1);
+              }
+
+              console.log(`build.yml for ${tag} is green (run ${run.id}). Deploying ${version}.`);
+              const before = await deployRunIds();
+
+              // Dispatch against the TAG, not master: the deploy applies the
+              // compose files under deploy/galactus/ from whatever ref it runs
+              // on, and those must be the ones this release was cut with.
+              const res = await fetch(
+                `${base}/actions/workflows/deploy-galactus.yml/dispatches`,
+                {
+                  method: "POST",
+                  headers: { ...headers, "Content-Type": "application/json" },
+                  body: JSON.stringify({
+                    ref: tagRef,
+                    inputs: {
+                      tag: version,
+                      scope: "app",
+                      bootstrap: "false",
+                      skip_migrate: "false",
+                    },
+                  }),
+                },
+              );
+              if (!res.ok) {
+                console.log(`::error::Dispatch returned HTTP ${res.status}: ${await res.text()}`);
+                console.log(`::error::Images for ${version} are published. Run`);
+                console.log(`::error::"Deploy to galactus" by hand with tag=${version}.`);
+                process.exit(1);
+              }
+
+              // A 204 only means Gitea accepted the request. Confirm a NEW run
+              // exists — an accepted call that creates no run is the failure mode
+              // that cost v1.0.3 its images.
+              for (let i = 0; i < 3; i++) {
+                await sleep(5_000);
+                const fresh = [...(await deployRunIds())].filter((id) => !before.has(id));
+                if (fresh.length) {
+                  console.log(`Deploy of ${version} to galactus is running (run ${fresh[0]}).`);
+                  return;
+                }
+              }
+
+              console.log(`::error::Dispatch was accepted but no deploy run appeared.`);
+              console.log(`::error::Run "Deploy to galactus" by hand with tag=${version}.`);
+              process.exit(1);
+            })();
+          '

.gitea/workflows/release.yml

@@ -1,10 +1,16 @@
 # Cut a release: stamp the version across every package.json, commit, tag, push.
 #
-# This does NOT build and does NOT deploy. Pushing the `vX.Y.Z` tag is what
+# This does NOT build and does NOT deploy itself. Pushing the `vX.Y.Z` tag is
-# triggers build.yml, which publishes `X.Y.Z`, `X.Y`, `sha-<short>` and `latest`
+# what triggers both build.yml, which publishes the `X.Y.Z`, `X.Y`,
-# image tags. Deploying stays a separate, deliberate act: once the build is
+# `sha-<short>` and `latest` image tags, and deploy-on-tag.yml, which waits for
-# green, dispatch deploy-galactus.yml with `tag=X.Y.Z` (no leading v — the tag
+# that build to go green and then dispatches deploy-galactus.yml with
-# carries the `v`, the image tag does not).
+# `tag=X.Y.Z scope=app` (no leading v — the git tag carries the `v`, the image
+# tag does not). A tag is the only ref that starts either chain; pushing to
+# master builds images and stops there.
+#
+# So cutting a release DOES reach prod. To cut a version without deploying it,
+# set the repo variable AUTO_DEPLOY_GALACTUS=false first; deploy-on-tag.yml then
+# prints the manual command instead of running it.
 #
 # Why a workflow instead of three local commands: the release commit is the one
 # thing that must be identical every time, and cutting it from a laptop is how
@@ -266,5 +272,9 @@ jobs:
           echo "Released v${VERSION}."
           echo ""
           echo "build.yml is now building git.mancinas.io/rmancinas/jorgecuadros-{api,web}:${VERSION}."
-          echo "When it is green, dispatch 'Deploy to galactus' with:"
+          echo "deploy-on-tag.yml is watching that build; when it goes green it dispatches"
-          echo "  tag=${VERSION}  scope=app  bootstrap=false  skip_migrate=false"
+          echo "'Deploy to galactus' with tag=${VERSION} scope=app bootstrap=false skip_migrate=false."
+          echo ""
+          echo "Watch that run. If it did not start (or AUTO_DEPLOY_GALACTUS=false),"
+          echo "dispatch 'Deploy to galactus' by hand with the same inputs."
+          echo "Rollback = re-dispatch it with an older tag."

apps/api/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@jorgecuadros/api",
-  "version": "1.0.7",
+  "version": "1.0.15",
   "private": true,
   "scripts": {
     "build": "nest build",

apps/api/src/billing/balance-floor.spec.ts

@@ -0,0 +1,180 @@
+import { Prisma } from "@jorgecuadros/database";
+import {
+  BALANCE_FLOOR_JOIN,
+  BALANCE_FORWARD_TYPE,
+  BillingService,
+  NOT_SUPERSEDED,
+} from "./billing.service";
+
+/**
+ * The balance floor drops rows a later BALANCE FORWARD already accounts for.
+ *
+ * It is worth testing because it fails silently: nothing throws, the numbers are
+ * just wrong, and they were wrong for years — the whole book read +20.6M MXN in
+ * credit because every customer's pre-cutover history was counted twice, once
+ * inside their opening balance and once as itself.
+ */
+describe("balance floor", () => {
+  describe("SQL fragments", () => {
+    it("binds the type name rather than interpolating it", () => {
+      // A literal would be a second place to edit if the label ever changes,
+      // and this string reaches SQL from a module constant.
+      expect(BALANCE_FLOOR_JOIN.values).toEqual([BALANCE_FORWARD_TYPE]);
+    });
+
+    it("keys the floor to the row's own customer", () => {
+      // Without this the derived table cross-joins and every customer inherits
+      // the earliest BALANCE FORWARD in the book.
+      expect(BALANCE_FLOOR_JOIN.sql).toContain(
+        "bfloor ON bfloor.customerId = t.customerId",
+      );
+    });
+
+    it("takes the most recent opening balance, not the first", () => {
+      // A customer accumulates one BALANCE FORWARD per year. MIN would floor at
+      // the oldest and leave every intervening year double-counted.
+      expect(BALANCE_FLOOR_JOIN.sql).toContain("MAX(bf.transactionDate)");
+      expect(BALANCE_FLOOR_JOIN.sql).not.toContain("MIN(bf.transactionDate)");
+    });
+
+    it("ignores voided opening balances when locating the floor", () => {
+      expect(BALANCE_FLOOR_JOIN.sql).toContain("bf.voidedAt IS NULL");
+    });
+
+    it("is inclusive of the opening balance row itself", () => {
+      // `>` instead of `>=` would drop the carried balance and understate every
+      // customer by exactly that amount.
+      expect(NOT_SUPERSEDED.sql).toContain("t.transactionDate >= bfloor.floorDate");
+      expect(NOT_SUPERSEDED.sql).not.toMatch(/transactionDate\s*>\s*bfloor/);
+    });
+
+    it("leaves customers with no opening balance untouched", () => {
+      // NULL comparisons are never true, so without the explicit IS NULL branch
+      // a customer who has no BALANCE FORWARD row loses their entire ledger.
+      expect(NOT_SUPERSEDED.sql).toContain("bfloor.floorDate IS NULL");
+    });
+
+    it("only ever references the alias the join defines", () => {
+      // The predicate is useless without the join; pairing them wrongly is a
+      // runtime "unknown column", so keep the alias identical in both.
+      const aliases = NOT_SUPERSEDED.sql.match(/bfloor\.\w+/g) ?? [];
+      expect(aliases.length).toBeGreaterThan(0);
+      for (const ref of aliases) {
+        expect(BALANCE_FLOOR_JOIN.sql).toContain(ref.split(".")[1]);
+      }
+    });
+  });
+
+  describe("statement()", () => {
+    /**
+     * One customer means one floor date, so the statement uses a scalar lookup
+     * instead of the join. Asserting on the `where` Prisma is handed is the only
+     * way to see it without a database.
+     */
+    function serviceWith(floor: Date | null) {
+      const findMany = jest.fn().mockResolvedValue([]);
+      const prisma = {
+        customer: {
+          findUnique: jest.fn().mockResolvedValue({
+            id: "c1",
+            name: "CUADROS, JORGE H.",
+            preferredCurrency: "USD",
+            _count: { properties: 0, policies: 0 },
+          }),
+        },
+        transaction: {
+          findFirst: jest
+            .fn()
+            .mockResolvedValue(floor ? { transactionDate: floor } : null),
+          findMany,
+        },
+      };
+
+      return {
+        service: new BillingService(prisma as never),
+        prisma,
+        findMany,
+      };
+    }
+
+    it("looks the floor up from the customer's newest opening balance", async () => {
+      const { service, prisma } = serviceWith(new Date("2026-01-01T00:00:00Z"));
+
+      await service.statement("c1");
+
+      expect(prisma.transaction.findFirst).toHaveBeenCalledWith(
+        expect.objectContaining({
+          where: {
+            customerId: "c1",
+            voidedAt: null,
+            type: { nameEn: BALANCE_FORWARD_TYPE },
+          },
+          orderBy: { transactionDate: "desc" },
+          select: { transactionDate: true },
+        }),
+      );
+    });
+
+    it("bounds the statement at the floor, inclusive", async () => {
+      const floor = new Date("2026-01-01T00:00:00Z");
+      const { service, findMany } = serviceWith(floor);
+
+      await service.statement("c1");
+
+      expect(findMany.mock.calls[0][0].where).toMatchObject({
+        customerId: "c1",
+        transactionDate: { gte: floor },
+      });
+    });
+
+    it("applies no date bound when the customer has no opening balance", async () => {
+      const { service, findMany } = serviceWith(null);
+
+      await service.statement("c1");
+
+      expect(findMany.mock.calls[0][0].where).not.toHaveProperty(
+        "transactionDate",
+      );
+    });
+
+    it("keeps the source-table exclusion alongside the floor", async () => {
+      // The two guards answer different questions — one reproduces legacy's
+      // DATOS2-only materialization, the other drops superseded history — and
+      // dropping either one changes the customer's balance.
+      const { service, findMany } = serviceWith(new Date("2026-01-01T00:00:00Z"));
+
+      await service.statement("c1");
+
+      const where = findMany.mock.calls[0][0].where;
+      expect(where.OR).toEqual([
+        { legacySourceTable: null },
+        { legacySourceTable: { notIn: expect.arrayContaining(["EFECTIVO"]) } },
+      ]);
+    });
+  });
+
+  describe("regression: NUMid 501", () => {
+    /**
+     * The arithmetic that exposed the bug, pinned so it cannot silently return.
+     * Figures measured against the live ledger on 2026-08-05.
+     */
+    const openingBalance = new Prisma.Decimal("-6732.29");
+    const activitySinceOpening = new Prisma.Decimal("-7333.00");
+    const preCutoverCashAlreadyInOpening = new Prisma.Decimal("3596.00");
+
+    it("matches the legacy portal once superseded rows are dropped", () => {
+      expect(openingBalance.plus(activitySinceOpening).toFixed(2)).toBe(
+        "-14065.29",
+      );
+    });
+
+    it("reproduces the wrong figure when they are not", () => {
+      expect(
+        openingBalance
+          .plus(activitySinceOpening)
+          .plus(preCutoverCashAlreadyInOpening)
+          .toFixed(2),
+      ).toBe("-10469.29");
+    });
+  });
+});

apps/api/src/billing/billing.service.ts

@@ -104,24 +104,31 @@ interface BalanceRow {
   nameMissing: number;
   city: string | null;
   state: string | null;
-  movements: bigint | number | string;
+  movements: RawCount;
   balanceMxn: Prisma.Decimal | null;
   balanceUsd: Prisma.Decimal | null;
   chargesMxn: Prisma.Decimal | null;
   creditsMxn: Prisma.Decimal | null;
   chargesUsd: Prisma.Decimal | null;
   creditsUsd: Prisma.Decimal | null;
-  utilityMovements: bigint | number | string;
+  utilityMovements: RawCount;
-  insuranceMovements: bigint | number | string;
+  insuranceMovements: RawCount;
   lastMovement: Date | null;
 }
 
 /**
- * Raw-query counts come back in three shapes depending on the aggregate:
+ * Every shape a raw-query count can arrive in. `COUNT(*)` is a bigint,
- * `COUNT(*)` as bigint, `SUM(bool)` as a decimal *string*, and plain numbers.
+ * `SUM(bool)` is a Prisma.Decimal, and plain numbers occur too — none of which
- * Normalize all of them before they reach the client as JSON.
+ * survive JSON serialization the way the client expects.
  */
-function num(v: bigint | number | string | null | undefined): number {
+type RawCount = bigint | number | string | Prisma.Decimal;
+
+/**
+ * Normalizes a raw-query count before it reaches the client as JSON. A bigint
+ * throws on JSON.stringify and a Decimal serializes to a *string*, so counts
+ * must not be passed through untouched.
+ */
+function num(v: RawCount | null | undefined): number {
   if (v === null || v === undefined) return 0;
   return typeof v === "number" ? v : Number(v);
 }
@@ -152,6 +159,56 @@ const NOT_VOIDED: Prisma.TransactionWhereInput = { voidedAt: null };
  */
 const NOT_OUTSTANDING: Prisma.TransactionWhereInput = { outstanding: false };
 
+/**
+ * The legacy type name for a carried-forward opening balance.
+ *
+ * These rows are not movements. Access materialized one per customer per year,
+ * dated Jan 1, holding the closing balance of everything before it — that is
+ * what let the portal keep each year in its own table (`datosfreak` = current,
+ * `2025`, `2024`, ...) and still show a correct running balance from a single
+ * year's rows.
+ */
+export const BALANCE_FORWARD_TYPE = "BALANCE FORWARD";
+
+/**
+ * Per-customer date of the most recent BALANCE FORWARD row.
+ *
+ * Joined rather than correlated: one small derived table (1,170 rows) beats a
+ * subquery evaluated per ledger row.
+ */
+export const BALANCE_FLOOR_JOIN = Prisma.sql`
+  LEFT JOIN (
+    SELECT bf.customerId, MAX(bf.transactionDate) AS floorDate
+    FROM transactions bf
+    JOIN type_transactions bft ON bft.id = bf.typeId
+    WHERE bft.nameEn = ${BALANCE_FORWARD_TYPE} AND bf.voidedAt IS NULL
+    GROUP BY bf.customerId
+  ) bfloor ON bfloor.customerId = t.customerId`;
+
+/**
+ * Excludes rows a later BALANCE FORWARD already accounts for.
+ *
+ * WHY THIS EXISTS. The platform holds both the synthetic BALANCE FORWARD rows
+ * and the real pre-cutover history they summarize, so summing a customer's
+ * whole ledger counts that history twice — once inside the opening balance,
+ * once as itself. NUMid 501 read -10,469.29 on the worklist against -14,065.29
+ * on the customer's own statement and on the legacy portal, the gap being two
+ * cash receipts from 2009 and 2012 that the 2026 opening balance had already
+ * absorbed.
+ *
+ * The scale is what settles it: summed the old way the entire book came to
+ * +20,605,447.86 MXN — the office owing its customers 20.6 million pesos.
+ * Floored, it is -56,855.90, a modest net receivable. A receivables ledger
+ * cannot be 20M in credit.
+ *
+ * Applies to BALANCES ONLY, in the same spirit as NOT_OUTSTANDING: the movement
+ * browser still totals every captured row, because "how much water did we
+ * capture in April" is a question about what was recorded, not about what is
+ * owed. Customers with no BALANCE FORWARD row (the floor is NULL) are
+ * unaffected.
+ */
+export const NOT_SUPERSEDED = Prisma.sql`(bfloor.floorDate IS NULL OR t.transactionDate >= bfloor.floorDate)`;
+
 /**
  * Source tables excluded from the customer-facing statement.
  *
@@ -402,22 +459,24 @@ export class BillingService {
         MAX(t.transactionDate)                                                     AS lastMovement
       FROM customers c
       JOIN transactions t ON t.customerId = c.id
-      WHERE t.voidedAt IS NULL AND t.outstanding = 0 ${nameFilter} ${txFilter}
+      ${BALANCE_FLOOR_JOIN}
+      WHERE t.voidedAt IS NULL AND t.outstanding = 0 AND ${NOT_SUPERSEDED} ${nameFilter} ${txFilter}
       GROUP BY c.id, c.name, c.nameSource, c.nameMissing, c.city, c.state
       ${having}
       ${orderBy}
       LIMIT ${pageSize} OFFSET ${(page - 1) * pageSize}
     `;
 
-    const counted = await this.prisma.$queryRaw<{ total: bigint | number | string }[]>`
+    const counted = await this.prisma.$queryRaw<{ total: RawCount }[]>`
       SELECT COUNT(*) AS total FROM (
         SELECT c.id
         FROM customers c
         JOIN transactions t ON t.customerId = c.id
+        ${BALANCE_FLOOR_JOIN}
         -- Must match the page query's filters exactly, or the total disagrees
         -- with the rows. (The void exclusion was missing here before the
         -- outstanding work; a voided-only customer inflated the count.)
-        WHERE t.voidedAt IS NULL AND t.outstanding = 0 ${nameFilter} ${txFilter}
+        WHERE t.voidedAt IS NULL AND t.outstanding = 0 AND ${NOT_SUPERSEDED} ${nameFilter} ${txFilter}
         GROUP BY c.id
         ${having}
       ) x
@@ -458,9 +517,18 @@ export class BillingService {
     };
   }
 
-  /** Top-line figures for the billing page header. */
+  /**
+   * Top-line figures for the billing page header.
+   *
+   * Two different questions live here and they use different row sets.
+   * `movements`, `ledgerCustomers`, `crossLineCustomers` and the date range are
+   * INVENTORY — what is stored — and count everything not voided. Everything
+   * under `byCurrency` / `byDomain` is a BALANCE, so it applies NOT_SUPERSEDED
+   * and drops rows an opening balance already accounts for. The four aggregates
+   * moved from Prisma groupBy to raw SQL to express that join; groupBy cannot.
+   */
   async stats() {
-    const [movements, ledgerCustomers, byCurrency, byDomain] = await Promise.all([
+    const [movements, ledgerCustomers] = await Promise.all([
       this.prisma.transaction.count({ where: NOT_VOIDED }),
       this.prisma.transaction
         .findMany({
@@ -469,34 +537,47 @@ export class BillingService {
           select: { customerId: true },
         })
         .then((r) => r.length),
-      this.prisma.transaction.groupBy({
-        by: ["currency"],
-        where: NOT_VOIDED,
-        _sum: { amount: true },
-        _count: { _all: true },
-      }),
-      this.prisma.transaction.groupBy({
-        by: ["domain", "currency"],
-        where: NOT_VOIDED,
-        _sum: { amount: true },
-        _count: { _all: true },
-      }),
     ]);
 
-    const charges = await this.prisma.transaction.groupBy({
+    const byCurrency = await this.prisma.$queryRaw<
-      by: ["currency"],
+      {
-      where: { AND: [{ amount: { lt: 0 } }, NOT_VOIDED] },
+        currency: string;
-      _sum: { amount: true },
+        net: Prisma.Decimal | null;
-      _count: { _all: true },
+        count: RawCount;
-    });
+        charges: Prisma.Decimal | null;
-    const credits = await this.prisma.transaction.groupBy({
+        chargeCount: RawCount;
-      by: ["currency"],
+        credits: Prisma.Decimal | null;
-      where: { AND: [{ amount: { gt: 0 } }, NOT_VOIDED] },
+        creditCount: RawCount;
-      _sum: { amount: true },
+      }[]
-      _count: { _all: true },
+    >`
-    });
+      SELECT t.currency                                             AS currency,
-    const chargeMap = new Map(charges.map((c) => [c.currency, c]));
+             SUM(t.amount)                                          AS net,
-    const creditMap = new Map(credits.map((c) => [c.currency, c]));
+             COUNT(*)                                               AS count,
+             SUM(CASE WHEN t.amount < 0 THEN t.amount ELSE 0 END)   AS charges,
+             SUM(t.amount < 0)                                      AS chargeCount,
+             SUM(CASE WHEN t.amount > 0 THEN t.amount ELSE 0 END)   AS credits,
+             SUM(t.amount > 0)                                      AS creditCount
+      FROM transactions t
+      ${BALANCE_FLOOR_JOIN}
+      WHERE t.voidedAt IS NULL AND ${NOT_SUPERSEDED}
+      GROUP BY t.currency
+    `;
+
+    const byDomain = await this.prisma.$queryRaw<
+      {
+        domain: string;
+        currency: string;
+        net: Prisma.Decimal | null;
+        count: RawCount;
+      }[]
+    >`
+      SELECT t.domain AS domain, t.currency AS currency,
+             SUM(t.amount) AS net, COUNT(*) AS count
+      FROM transactions t
+      ${BALANCE_FLOOR_JOIN}
+      WHERE t.voidedAt IS NULL AND ${NOT_SUPERSEDED}
+      GROUP BY t.domain, t.currency
+    `;
 
     // How many customers sit on each side of the line, per currency — the
     // headline for a receivables view. Counted in SQL; a customer can be
@@ -504,16 +585,19 @@ export class BillingService {
     const sides = await this.prisma.$queryRaw<
       {
         currency: string;
-        owing: bigint | number | string;
+        owing: RawCount;
-        inCredit: bigint | number | string;
+        inCredit: RawCount;
       }[]
     >`
       SELECT currency,
              SUM(bal < -0.005) AS owing,
              SUM(bal >  0.005) AS inCredit
       FROM (
-        SELECT customerId, currency, SUM(amount) AS bal
+        SELECT t.customerId, t.currency, SUM(t.amount) AS bal
-        FROM transactions WHERE voidedAt IS NULL GROUP BY customerId, currency
+        FROM transactions t
+        ${BALANCE_FLOOR_JOIN}
+        WHERE t.voidedAt IS NULL AND ${NOT_SUPERSEDED}
+        GROUP BY t.customerId, t.currency
       ) x
       GROUP BY currency
     `;
@@ -534,7 +618,7 @@ export class BillingService {
 
     // Customers whose ledger spans both business lines — the whole reason this
     // module is one view instead of two.
-    const crossLine = await this.prisma.$queryRaw<{ n: bigint | number | string }[]>`
+    const crossLine = await this.prisma.$queryRaw<{ n: RawCount }[]>`
       SELECT COUNT(*) AS n FROM (
         SELECT customerId FROM transactions WHERE voidedAt IS NULL
         GROUP BY customerId HAVING COUNT(DISTINCT domain) > 1
@@ -549,20 +633,20 @@ export class BillingService {
       lastMovement: lastRow?.transactionDate ?? null,
       byCurrency: byCurrency.map((c) => ({
         currency: c.currency,
-        net: c._sum.amount,
+        net: c.net,
-        count: c._count._all,
+        count: num(c.count),
-        charges: chargeMap.get(c.currency)?._sum.amount ?? null,
+        charges: c.charges,
-        chargeCount: chargeMap.get(c.currency)?._count._all ?? 0,
+        chargeCount: num(c.chargeCount),
-        credits: creditMap.get(c.currency)?._sum.amount ?? null,
+        credits: c.credits,
-        creditCount: creditMap.get(c.currency)?._count._all ?? 0,
+        creditCount: num(c.creditCount),
         owing: num(sideMap.get(c.currency)?.owing),
         inCredit: num(sideMap.get(c.currency)?.inCredit),
       })),
       byDomain: byDomain.map((d) => ({
         domain: d.domain,
         currency: d.currency,
-        net: d._sum.amount,
+        net: d.net,
-        count: d._count._all,
+        count: num(d.count),
       })),
     };
   }
@@ -589,7 +673,7 @@ export class BillingService {
     });
 
     const years = await this.prisma.$queryRaw<
-      { year: number; count: bigint | number | string }[]
+      { year: number; count: RawCount }[]
     >`
       SELECT YEAR(transactionDate) AS year, COUNT(*) AS count
       FROM transactions WHERE voidedAt IS NULL GROUP BY year ORDER BY year DESC
@@ -647,9 +731,32 @@ export class BillingService {
       throw new NotFoundException(`Customer ${customerId} not found`);
     }
 
+    // One customer, so the balance floor is a single date rather than the
+    // derived table the aggregate queries join. See NOT_SUPERSEDED: rows before
+    // the opening balance are already inside it, and showing them would both
+    // double the total and make every balanceAfter below wrong.
+    //
+    // This is also what stops FEE ANUAL and fee15 leaking in. They are not in
+    // STATEMENT_EXCLUDED_SOURCE_TABLES — that list exists to reproduce legacy's
+    // DATOS2-only `datosfreak`, and it was letting 2,092 pre-cutover fee rows
+    // across 1,062 customers through, skewing the statement by -5,129,764
+    // against the number those customers have been quoted for years. Dating
+    // rather than source is the right test: a FEE ANUAL row *after* the opening
+    // balance is a real charge and still counts.
+    const floor = await this.prisma.transaction.findFirst({
+      where: {
+        customerId,
+        voidedAt: null,
+        type: { nameEn: BALANCE_FORWARD_TYPE },
+      },
+      orderBy: { transactionDate: "desc" },
+      select: { transactionDate: true },
+    });
+
     const rows = await this.prisma.transaction.findMany({
       where: {
         customerId,
+        ...(floor ? { transactionDate: { gte: floor.transactionDate } } : {}),
         // NULL-safe exclusion. `notIn` alone compiles to SQL `NOT IN`, and
         // `NULL NOT IN (...)` is NULL, not true — so every app-captured row
         // (which has no legacySourceTable) silently vanished from the

apps/api/src/ops/job-progress.spec.ts

@@ -0,0 +1,76 @@
+import { jobProgress } from "./ops.service";
+
+/** Shape run_all.py emits, with the shell trace lines it interleaves. */
+const line = (i: number, n: number, name: string) =>
+  `[paso ${i}/${n}] ${name}\n+ /repo/migration/.venv/bin/python /repo/migration/${name} --env prod\n[${name}] target env: prod\n  validation: OK`;
+
+describe("jobProgress", () => {
+  it("returns null before any step marker appears", () => {
+    // The safety backup runs before run_all.py, so this is the real state for
+    // the first stretch of every REIMPORT.
+    expect(jobProgress("== Respaldo de seguridad previo ==\ntablas capturadas: 39", "RUNNING")).toBeNull();
+  });
+
+  it("returns null for jobs that have no steps at all", () => {
+    // BACKUP/RESTORE are a single mysqldump; a fabricated percentage would be
+    // worse than none.
+    expect(jobProgress("mysqldump ... done", "SUCCESS")).toBeNull();
+  });
+
+  it("tracks the most recent marker, not the first", () => {
+    const log = [line(1, 9, "transform_customers.py"), line(2, 9, "transform_properties.py")].join("\n");
+    const p = jobProgress(log, "RUNNING");
+    expect(p).toMatchObject({ step: 2, total: 9, name: "transform_properties.py" });
+  });
+
+  /**
+   * The point of the whole feature. While RUNNING, step i is IN PROGRESS, so
+   * only i-1 are done. Counting i as complete would show 100% while the final
+   * and slowest step (blob_extract) is still working.
+   */
+  it("does not claim a running step is finished", () => {
+    expect(jobProgress(line(1, 9, "transform_customers.py"), "RUNNING")?.percent).toBe(0);
+    expect(jobProgress(line(9, 9, "blob_extract.py"), "RUNNING")?.percent).toBe(88);
+  });
+
+  it("reaches 100 only once the job is no longer running", () => {
+    expect(jobProgress(line(9, 9, "blob_extract.py"), "SUCCESS")?.percent).toBe(100);
+  });
+
+  /** A job that died mid-way must report where it died, not 100%. */
+  it("reports the failed step rather than completion", () => {
+    const p = jobProgress(line(5, 9, "transform_transactions.py"), "FAILED");
+    expect(p).toMatchObject({ step: 5, total: 9 });
+    expect(p!.percent).toBe(55);
+  });
+
+  it("handles the 8-step SYNC list as well as the 9-step REIMPORT one", () => {
+    expect(jobProgress(line(8, 8, "transform_bank.py"), "SUCCESS")?.percent).toBe(100);
+    expect(jobProgress(line(4, 8, "transform_policies.py"), "RUNNING")?.percent).toBe(37);
+  });
+
+  /**
+   * Captured verbatim from `run_all.run(..., step=8, total=9)`. This is the
+   * contract between the Python and this parser; if run_all.py's format
+   * changes, this fails rather than the panel silently showing no progress.
+   */
+  it("parses the exact line run_all.py emits", () => {
+    const real =
+      "[paso 8/9] transform_bank.py\n+ /repo/migration/.venv/bin/python /repo/migration/transform_bank.py --env prod";
+    expect(jobProgress(real, "RUNNING")).toMatchObject({
+      step: 8,
+      total: 9,
+      name: "transform_bank.py",
+      percent: 77,
+    });
+  });
+
+  it("ignores a malformed marker instead of reporting NaN", () => {
+    expect(jobProgress("[paso 3/0] x.py", "RUNNING")).toBeNull();
+  });
+
+  /** The marker must be at line start so log text quoting it cannot spoof it. */
+  it("does not match a marker embedded mid-line", () => {
+    expect(jobProgress("some output mentioning [paso 4/9] fake.py", "RUNNING")).toBeNull();
+  });
+});

apps/api/src/ops/ops.controller.ts

@@ -19,6 +19,7 @@ import { AbilityGuard } from "../auth/ability.guard";
 import { RequireAbility } from "../auth/require-ability.decorator";
 import { AuditService } from "../common/audit.service";
 import { OpsService } from "./ops.service";
+import { ReplicationService } from "./replication.service";
 import { StartJobDto } from "./start-job.dto";
 
 /** Every route is ADMIN-only (ability "db:manage"). */
@@ -28,6 +29,7 @@ import { StartJobDto } from "./start-job.dto";
 export class OpsController {
   constructor(
     private readonly ops: OpsService,
+    private readonly replication: ReplicationService,
     private readonly audit: AuditService,
   ) {}
 
@@ -96,6 +98,12 @@ export class OpsController {
 
   /* --------------------------------------------------------------- jobs */
 
+  /** Health of the my.jorgecuadros.com read replica. Read-only, no audit entry. */
+  @Get("replication")
+  replicationStatus() {
+    return this.replication.status();
+  }
+
   @Get("jobs")
   listJobs() {
     return this.ops.listJobs();

apps/api/src/ops/ops.module.ts

@@ -1,9 +1,10 @@
 import { Module } from "@nestjs/common";
 import { OpsController } from "./ops.controller";
 import { OpsService } from "./ops.service";
+import { ReplicationService } from "./replication.service";
 
 @Module({
   controllers: [OpsController],
-  providers: [OpsService],
+  providers: [OpsService, ReplicationService],
 })
 export class OpsModule {}

apps/api/src/ops/ops.service.ts

@@ -67,6 +67,55 @@ export class OpsService implements OnModuleInit {
   async onModuleInit(): Promise<void> {
     await fs.mkdir(this.ingestDir, { recursive: true });
     await fs.mkdir(this.backupDir, { recursive: true });
+    await this.reconcileOrphanedJobs();
+  }
+
+  /**
+   * Fail any job still marked RUNNING at startup.
+   *
+   * Jobs run as a child of THIS process, so no job can outlive it: if a row says
+   * RUNNING while we are booting, its process died with the previous instance
+   * and nothing will ever finalize it. Since startJob() refuses to start while
+   * any RUNNING row exists, one interrupted job wedges the panel permanently
+   * with no way out from the UI — it took a manual UPDATE against production to
+   * recover the first time this happened, when a deploy landed 110 seconds into
+   * a REIMPORT.
+   *
+   * Deliberately unconditional rather than filtered on age: "started recently"
+   * does not mean "still alive" here, and a fresh boot is proof enough that
+   * nothing survived.
+   */
+  private async reconcileOrphanedJobs(): Promise<void> {
+    try {
+      // Read then write one by one rather than updateMany: the log needs the
+      // reason APPENDED, and a job whose log just stops mid-step with no
+      // explanation is what made the first occurrence hard to diagnose.
+      const orphans = await this.prisma.opsJob.findMany({
+        where: { status: "RUNNING" },
+        select: { id: true, kind: true, log: true },
+      });
+      for (const job of orphans) {
+        await this.prisma.opsJob.update({
+          where: { id: job.id },
+          data: {
+            status: "FAILED",
+            finishedAt: new Date(),
+            log: {
+              set:
+                job.log +
+                "\n[interrumpido: el contenedor se reinició mientras el trabajo corría; " +
+                "el proceso hijo no sobrevive a un redespliegue. " +
+                "Vuelva a ejecutar la operación desde el principio.]\n",
+            },
+          },
+        });
+        this.logger.warn(`trabajo ${job.kind} ${job.id} quedó huérfano; marcado FAILED`);
+      }
+    } catch (e) {
+      // Never block startup on this. A failed reconcile leaves the panel
+      // wedged, which is bad, but an API that will not boot is worse.
+      this.logger.error(`no se pudieron reconciliar trabajos huérfanos: ${String(e)}`);
+    }
   }
 
   /* -------------------------------------------------------------- ingest */
@@ -163,7 +212,9 @@ export class OpsService implements OnModuleInit {
   async getJob(id: string) {
     const job = await this.prisma.opsJob.findUnique({ where: { id } });
     if (!job) throw new NotFoundException("Trabajo no encontrado.");
-    return job;
+    // Derived, never stored: the log is the single source of truth for how far
+    // a job got, so progress cannot drift out of sync with it.
+    return { ...job, progress: jobProgress(job.log, job.status) };
   }
 
   /**
@@ -349,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 } };
     }
 
@@ -465,3 +521,52 @@ export class OpsService implements OnModuleInit {
 function shq(v: string): string {
   return `'${v.replace(/'/g, `'\\''`)}'`;
 }
+
+/** Progress derived from a job's log. Null when the job reports no steps. */
+export interface JobProgress {
+  /** 1-based index of the step currently running (or last reached). */
+  step: number;
+  total: number;
+  /** Script name, e.g. "transform_bank.py". */
+  name: string;
+  /** 0..100, floored. 100 only once the job is no longer RUNNING. */
+  percent: number;
+}
+
+/**
+ * Parse the "[paso i/N] name" markers migration/run_all.py emits.
+ *
+ * Progress is DERIVED from the log rather than tracked in a column: the log is
+ * already the record of what happened, and a separate counter could disagree
+ * with it — which is exactly the confusion a progress display is supposed to
+ * remove. run_all.py owns the step count, so adding a step cannot desync this.
+ *
+ * BACKUP and RESTORE are a single mysqldump with no steps, so they return null
+ * and the UI shows an indeterminate spinner. Reporting a fabricated percentage
+ * for them would be worse than showing none.
+ */
+export function jobProgress(
+  log: string,
+  status: string,
+): JobProgress | null {
+  // Last marker wins: the log grows, and the newest line is the current step.
+  const matches = [...log.matchAll(/^\[paso (\d+)\/(\d+)\] (\S+)/gm)];
+  const last = matches[matches.length - 1];
+  if (!last) return null;
+
+  const step = Number(last[1]);
+  const total = Number(last[2]);
+  if (!Number.isFinite(step) || !Number.isFinite(total) || total <= 0) return null;
+
+  // While RUNNING, step i means i is IN PROGRESS, not finished — so report
+  // (i-1) completed. Claiming 100% while the last step is still working is the
+  // classic progress-bar lie, and here the last step (blob_extract) is also the
+  // slowest, so it would sit at "100%" for the longest stretch of the job.
+  const done = status === "RUNNING" ? step - 1 : step;
+  return {
+    step,
+    total,
+    name: last[3],
+    percent: Math.max(0, Math.min(100, Math.floor((done / total) * 100))),
+  };
+}

apps/api/src/ops/replication.service.ts

@@ -0,0 +1,261 @@
+import { Injectable, Logger } from "@nestjs/common";
+import { execFile } from "node:child_process";
+import { promisify } from "node:util";
+
+const exec = promisify(execFile);
+
+/**
+ * How far the SQL thread is behind the I/O thread, in source binlog bytes.
+ *
+ * This is a different question from `secondsBehind`, and it answers the case
+ * that lag hides: while the SQL thread grinds through one huge transaction,
+ * `Seconds_Behind_Source` can sit still or even read 0, but the relay backlog
+ * is plainly shrinking (or not). It costs nothing extra — every field here
+ * comes out of the same `SHOW REPLICA STATUS` the panel already runs.
+ *
+ * Both positions are coordinates in the SOURCE's binlog, so they are only
+ * comparable while both threads are working on the SAME source file. When they
+ * are not, the replica is whole files behind and the byte delta is meaningless
+ * (positions restart at ~4 in each new file), so `backlogBytes` and `percent`
+ * are null and `sameFile` says why.
+ */
+export interface ApplyProgress {
+  /** Source binlog file the I/O thread is currently reading. */
+  sourceLogFile: string | null;
+  /** Position in `sourceLogFile` that the I/O thread has fetched up to. */
+  readPos: number;
+  /** Source binlog file the SQL thread is currently applying. */
+  relayLogFile: string | null;
+  /** Position in `relayLogFile` that the SQL thread has applied up to. */
+  execPos: number;
+  /** True while both threads are on the same source file. */
+  sameFile: boolean;
+  /** Fetched-but-not-yet-applied bytes. Null when the files differ. */
+  backlogBytes: number | null;
+  /**
+   * `execPos / readPos` as a percentage, null when the files differ.
+   *
+   * Deliberately never rounded up to 100 while any backlog remains: binlog
+   * positions are large, so a real backlog of a few KB is 99.99% of the file
+   * and would render as "caught up" when it is not. Read `backlogBytes === 0`
+   * for actually caught up.
+   */
+  percent: number | null;
+}
+
+export interface ReplicationStatus {
+  /** false when the replica is not configured for this environment at all. */
+  configured: boolean;
+  /** true only when both threads run, no error is set, and lag is within bounds. */
+  healthy: boolean;
+  host: string | null;
+  ioRunning: string | null;
+  sqlRunning: string | null;
+  /** null when MySQL reports NULL, which it does whenever a thread is down. */
+  secondsBehind: number | null;
+  lastIoError: string | null;
+  lastSqlError: string | null;
+  sourceHost: string | null;
+  /** Relay-log apply progress. Null when the status output has no positions. */
+  apply: ApplyProgress | null;
+  /** Human-readable reason when healthy is false. */
+  problem: string | null;
+  checkedAt: string;
+}
+
+/**
+ * Reports whether the my.jorgecuadros.com read replica is still replicating.
+ *
+ * The replica is what the public site reads once the platformDataSource flag is
+ * on, and a replica that has silently stopped applying serves stale balances
+ * rather than erroring — the failure is invisible from the site itself, which is
+ * why it needs a panel.
+ *
+ * Shells out to the mysql client for the same reason the rest of OpsService
+ * does: there is no MySQL driver in this API's dependencies, and the image
+ * already ships one.
+ */
+@Injectable()
+export class ReplicationService {
+  private readonly logger = new Logger(ReplicationService.name);
+
+  /** Lag above this many seconds is reported as unhealthy. */
+  private readonly maxLagSeconds = Number(process.env.REPLICA_MAX_LAG ?? 60);
+
+  async status(): Promise<ReplicationStatus> {
+    const host = process.env.REPLICA_DB_HOST;
+    const user = process.env.REPLICA_DB_USER;
+    const password = process.env.REPLICA_DB_PASS;
+    const now = new Date().toISOString();
+
+    const empty: ReplicationStatus = {
+      configured: false,
+      healthy: false,
+      host: host ?? null,
+      ioRunning: null,
+      sqlRunning: null,
+      secondsBehind: null,
+      lastIoError: null,
+      lastSqlError: null,
+      sourceHost: null,
+      apply: null,
+      problem: null,
+      checkedAt: now,
+    };
+
+    if (!host || !user || !password) {
+      return { ...empty, problem: "REPLICA_DB_* no configuradas" };
+    }
+
+    let raw: string;
+    try {
+      // --ssl is required: the replica sets require_secure_transport=ON.
+      //
+      // --ssl-verify-server-cert=0 is deliberate and is NOT the same trade-off
+      // the website makes. This hop never leaves Tailscale — the replica is
+      // reached on its CGNAT tailnet address and its firewall admits only this
+      // host — so WireGuard already authenticates the peer. The DreamHost leg
+      // crosses the public internet and therefore pins the CA instead. The
+      // client here is MariaDB's, which rejects our self-signed CA outright
+      // unless it is handed the CA file, which would mean shipping a cert into
+      // this image for a link that is already authenticated.
+      const { stdout } = await exec(
+        "mysql",
+        [
+          `--host=${host}`,
+          `--user=${user}`,
+          "--ssl",
+          "--ssl-verify-server-cert=0",
+          "--connect-timeout=5",
+          "-e",
+          "SHOW REPLICA STATUS\\G",
+        ],
+        {
+          env: { ...process.env, MYSQL_PWD: password },
+          timeout: 15_000,
+        },
+      );
+      raw = stdout;
+    } catch (e) {
+      const msg = e instanceof Error ? e.message : String(e);
+      this.logger.warn(`no se pudo consultar la réplica: ${msg}`);
+      return { ...empty, configured: true, problem: `No se pudo conectar: ${msg}` };
+    }
+
+    const field = (name: string): string | null => replicaField(raw, name);
+
+    // An empty result set means the server is not configured as a replica at
+    // all — distinct from "configured but broken", and worth saying plainly.
+    if (!raw.includes("Replica_IO_Running")) {
+      return {
+        ...empty,
+        configured: true,
+        problem: "El servidor no está configurado como réplica",
+      };
+    }
+
+    const ioRunning = field("Replica_IO_Running");
+    const sqlRunning = field("Replica_SQL_Running");
+    const lagRaw = field("Seconds_Behind_Source");
+    const secondsBehind =
+      lagRaw === null || lagRaw === "NULL" ? null : Number(lagRaw);
+    const lastIoError = field("Last_IO_Error");
+    const lastSqlError = field("Last_SQL_Error");
+
+    // Order matters: report the most specific cause first. Checking lag before
+    // the threads would blame "sin dato de retraso" for what is really a
+    // stopped thread, because MySQL reports NULL lag whenever either is down.
+    let problem: string | null = null;
+    if (ioRunning !== "Yes") problem = "El hilo de E/S no está corriendo";
+    else if (sqlRunning !== "Yes") problem = "El hilo SQL no está corriendo";
+    else if (lastSqlError) problem = `Error SQL: ${lastSqlError}`;
+    else if (lastIoError) problem = `Error de E/S: ${lastIoError}`;
+    else if (secondsBehind === null) problem = "Sin dato de retraso";
+    else if (secondsBehind > this.maxLagSeconds)
+      problem = `Retraso de ${secondsBehind}s (máximo ${this.maxLagSeconds}s)`;
+
+    return {
+      configured: true,
+      healthy: problem === null,
+      host,
+      ioRunning,
+      sqlRunning,
+      secondsBehind,
+      lastIoError,
+      lastSqlError,
+      sourceHost: field("Source_Host"),
+      // Reported, never folded into `healthy`: a non-zero backlog is the normal
+      // state of a working replica for the instant between fetch and apply, so
+      // alarming on it would cry wolf. It is here to answer "is it moving?"
+      // when the lag counter is stuck.
+      apply: applyProgress(raw),
+      problem,
+      checkedAt: now,
+    };
+  }
+}
+
+/**
+ * Derive relay-apply progress from `SHOW REPLICA STATUS\G` output.
+ *
+ * Exported for testing. Free in query terms — it re-reads four more fields from
+ * the output the caller already has, with no second round trip to the replica
+ * and no connection to the source.
+ *
+ * @returns null when either position is missing or unparseable, which is what
+ *          happens on a server that is not a replica at all.
+ */
+export function applyProgress(raw: string): ApplyProgress | null {
+  const num = (name: string): number | null => {
+    const v = replicaField(raw, name);
+    if (v === null || v === "NULL") return null;
+    const n = Number(v);
+    return Number.isFinite(n) ? n : null;
+  };
+
+  const readPos = num("Read_Source_Log_Pos");
+  const execPos = num("Exec_Source_Log_Pos");
+  if (readPos === null || execPos === null) return null;
+
+  const sourceLogFile = replicaField(raw, "Source_Log_File");
+  const relayLogFile = replicaField(raw, "Relay_Source_Log_File");
+  const sameFile =
+    sourceLogFile !== null && relayLogFile !== null && sourceLogFile === relayLogFile;
+
+  // Clamped at 0: the SQL thread cannot be ahead of the I/O thread, but the two
+  // fields are sampled independently, so a rotation racing this read can print
+  // a momentarily negative delta. Zero is the honest floor, not a bug.
+  const backlogBytes = sameFile ? Math.max(0, readPos - execPos) : null;
+
+  let percent: number | null = null;
+  if (backlogBytes !== null && readPos > 0) {
+    // Truncate rather than round, and hold short of 100 while bytes remain —
+    // see the doc on ApplyProgress.percent.
+    const p = Math.floor((execPos / readPos) * 10_000) / 100;
+    percent = backlogBytes === 0 ? 100 : Math.min(p, 99.99);
+  }
+
+  return { sourceLogFile, readPos, relayLogFile, execPos, sameFile, backlogBytes, percent };
+}
+
+/**
+ * Read one field out of `SHOW REPLICA STATUS\G` output.
+ *
+ * Exported for testing, and worth testing: the obvious regex is wrong.
+ * `\s` matches newlines in JavaScript, so `^\s*NAME:\s*(.*)$` lets the `\s*`
+ * after the colon swallow the line break of an EMPTY field and capture the
+ * following line instead. Last_SQL_Error is empty on a healthy replica, so that
+ * version reported the next line ("Replicate_Ignore_Server_Ids:") as a SQL
+ * error and rendered a perfectly healthy replica as broken.
+ *
+ * Hence `[^\S\n]` — horizontal whitespace only — on both sides of the name.
+ *
+ * @returns the trimmed value, or null when the field is absent OR empty. Empty
+ *          and absent mean the same thing to every caller here: MySQL prints
+ *          error fields as blank rather than omitting them.
+ */
+export function replicaField(raw: string, name: string): string | null {
+  const m = raw.match(new RegExp(`^[^\\S\\n]*${name}:[^\\S\\n]*(.*)$`, "m"));
+  const v = m?.[1]?.trim();
+  return v === undefined || v === "" ? null : v;
+}

apps/api/src/ops/replication.spec.ts

@@ -0,0 +1,180 @@
+import { applyProgress, replicaField } from "./replication.service";
+
+/**
+ * Verbatim shape of `SHOW REPLICA STATUS\G` from the live replica, trimmed to
+ * the fields the panel reads plus the neighbours that matter.
+ *
+ * The empty `Last_SQL_Error:` immediately followed by
+ * `Replicate_Ignore_Server_Ids:` is the whole point of the fixture — that exact
+ * adjacency is what the first implementation misread.
+ */
+const HEALTHY = [
+  "*************************** 1. row ***************************",
+  "             Replica_IO_State: Waiting for source to send event",
+  "                  Source_Host: 100.103.77.46",
+  "                  Source_User: repl",
+  "              Source_Log_File: binlog.000042",
+  "          Read_Source_Log_Pos: 194884231",
+  "        Relay_Source_Log_File: binlog.000042",
+  "          Exec_Source_Log_Pos: 194884231",
+  "           Replica_IO_Running: Yes",
+  "          Replica_SQL_Running: Yes",
+  "              Replicate_Do_DB: ",
+  "                   Last_Errno: 0",
+  "                   Last_Error: ",
+  "        Seconds_Behind_Source: 0",
+  "                Last_IO_Errno: 0",
+  "                Last_IO_Error: ",
+  "               Last_SQL_Errno: 0",
+  "               Last_SQL_Error: ",
+  "  Replicate_Ignore_Server_Ids: ",
+  "             Source_Server_Id: 1",
+].join("\n");
+
+const BROKEN = [
+  "           Replica_IO_Running: Yes",
+  "          Replica_SQL_Running: No",
+  "        Seconds_Behind_Source: NULL",
+  "                Last_IO_Error: ",
+  "               Last_SQL_Error: Could not execute Write_rows event on table jorgecuadros.customers",
+  "  Replicate_Ignore_Server_Ids: ",
+].join("\n");
+
+describe("replicaField", () => {
+  it("reads plain values", () => {
+    expect(replicaField(HEALTHY, "Replica_IO_Running")).toBe("Yes");
+    expect(replicaField(HEALTHY, "Replica_SQL_Running")).toBe("Yes");
+    expect(replicaField(HEALTHY, "Source_Host")).toBe("100.103.77.46");
+    expect(replicaField(HEALTHY, "Seconds_Behind_Source")).toBe("0");
+  });
+
+  /**
+   * The regression this file exists for. `\s` matches newlines in JavaScript,
+   * so `^\s*NAME:\s*(.*)$` walks past an empty field's line break and captures
+   * the NEXT line — turning a healthy replica into
+   * "Error SQL: Replicate_Ignore_Server_Ids:" in the admin panel.
+   */
+  it("returns null for an empty field instead of the following line", () => {
+    expect(replicaField(HEALTHY, "Last_SQL_Error")).toBeNull();
+    expect(replicaField(HEALTHY, "Last_IO_Error")).toBeNull();
+    expect(replicaField(HEALTHY, "Last_Error")).toBeNull();
+    expect(replicaField(HEALTHY, "Replicate_Do_DB")).toBeNull();
+    expect(replicaField(HEALTHY, "Replicate_Ignore_Server_Ids")).toBeNull();
+  });
+
+  it("still reads a real error when there is one", () => {
+    expect(replicaField(BROKEN, "Last_SQL_Error")).toBe(
+      "Could not execute Write_rows event on table jorgecuadros.customers",
+    );
+    expect(replicaField(BROKEN, "Replica_SQL_Running")).toBe("No");
+  });
+
+  /** NULL is a distinct state from empty and must survive as the literal. */
+  it("preserves the literal NULL that MySQL prints for unknown lag", () => {
+    expect(replicaField(BROKEN, "Seconds_Behind_Source")).toBe("NULL");
+  });
+
+  it("returns null for a field that is not present at all", () => {
+    expect(replicaField(HEALTHY, "Nonexistent_Field")).toBeNull();
+  });
+
+  /**
+   * Field names are matched at the start of a line. Without the line anchor,
+   * "Last_Error" would also match inside "Last_SQL_Error" and read the wrong
+   * value — the two carry different things and both feed the panel.
+   */
+  it("does not match a field name that is a suffix of another", () => {
+    const raw = "               Last_SQL_Error: boom\n                   Last_Error: ";
+    expect(replicaField(raw, "Last_Error")).toBeNull();
+    expect(replicaField(raw, "Last_SQL_Error")).toBe("boom");
+  });
+});
+
+/** Builds the four position fields the apply-progress reader cares about. */
+function positions(
+  sourceFile: string,
+  readPos: number | string,
+  relayFile: string,
+  execPos: number | string,
+): string {
+  return [
+    `              Source_Log_File: ${sourceFile}`,
+    `          Read_Source_Log_Pos: ${readPos}`,
+    `        Relay_Source_Log_File: ${relayFile}`,
+    `          Exec_Source_Log_Pos: ${execPos}`,
+  ].join("\n");
+}
+
+describe("applyProgress", () => {
+  it("reports zero backlog and 100% when both positions match", () => {
+    const p = applyProgress(HEALTHY)!;
+    expect(p.sameFile).toBe(true);
+    expect(p.sourceLogFile).toBe("binlog.000042");
+    expect(p.readPos).toBe(194884231);
+    expect(p.execPos).toBe(194884231);
+    expect(p.backlogBytes).toBe(0);
+    expect(p.percent).toBe(100);
+  });
+
+  it("reports the byte delta when the SQL thread trails inside one file", () => {
+    const p = applyProgress(positions("binlog.000042", 2_000_000, "binlog.000042", 1_500_000))!;
+    expect(p.backlogBytes).toBe(500_000);
+    expect(p.percent).toBe(75);
+  });
+
+  /**
+   * The reason the byte delta exists at all. `Seconds_Behind_Source` holds at 0
+   * while the SQL thread is mid-transaction, so the backlog is the only field
+   * that moves — and the only one that says the replica is not caught up.
+   */
+  it("shows a backlog even when the lag counter reads zero", () => {
+    const raw = [
+      "        Seconds_Behind_Source: 0",
+      positions("binlog.000042", 900, "binlog.000042", 400),
+    ].join("\n");
+    expect(replicaField(raw, "Seconds_Behind_Source")).toBe("0");
+    expect(applyProgress(raw)!.backlogBytes).toBe(500);
+  });
+
+  /**
+   * Positions restart near 4 in every new binlog file, so subtracting across
+   * files produces a number that is not a backlog — here it would be a large
+   * NEGATIVE one, which would render as "ahead of the source".
+   */
+  it("refuses to compare positions across different binlog files", () => {
+    const p = applyProgress(positions("binlog.000043", 500, "binlog.000042", 194_000_000))!;
+    expect(p.sameFile).toBe(false);
+    expect(p.backlogBytes).toBeNull();
+    expect(p.percent).toBeNull();
+    expect(p.sourceLogFile).toBe("binlog.000043");
+    expect(p.relayLogFile).toBe("binlog.000042");
+  });
+
+  /**
+   * Percent must not round up to 100 while bytes remain: binlog positions are
+   * large, so a genuine backlog is a rounding error away from the whole file
+   * and would otherwise render as "caught up" on a replica that is not.
+   */
+  it("stops short of 100% while any backlog remains", () => {
+    const p = applyProgress(positions("binlog.000042", 194_884_231, "binlog.000042", 194_884_230))!;
+    expect(p.backlogBytes).toBe(1);
+    expect(p.percent).toBe(99.99);
+  });
+
+  /** Sampled independently, so a rotation racing the read can invert them. */
+  it("clamps a momentarily negative delta to zero", () => {
+    const p = applyProgress(positions("binlog.000042", 400, "binlog.000042", 500))!;
+    expect(p.backlogBytes).toBe(0);
+    expect(p.percent).toBe(100);
+  });
+
+  it("returns null when the server is not a replica and prints no positions", () => {
+    expect(applyProgress("")).toBeNull();
+    expect(applyProgress(BROKEN)).toBeNull();
+  });
+
+  /** A stopped thread makes MySQL print NULL, which is not a position. */
+  it("returns null when a position is NULL", () => {
+    expect(applyProgress(positions("binlog.000042", "NULL", "binlog.000042", 400))).toBeNull();
+  });
+});

apps/web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@jorgecuadros/web",
-  "version": "1.0.7",
+  "version": "1.0.15",
   "private": true,
   "scripts": {
     "dev": "next dev -p 4500",

apps/web/src/app/operaciones/page.tsx

@@ -14,6 +14,7 @@ import {
   deleteBackup,
   deleteIngest,
   getOpsJob,
+  getReplicationStatus,
   listBackups,
   listIngest,
   listOpsJobs,
@@ -22,10 +23,12 @@ import {
 } from "@/lib/api";
 import type { UploadProgress } from "@/lib/api";
 import type {
+  ApplyProgress,
   BackupFile,
   IngestFile,
   OpsJob,
   OpsJobKind,
+  ReplicationStatus,
 } from "@/lib/types";
 
 const INGEST_MAX_BYTES = 2 * 1024 * 1024 * 1024;
@@ -223,10 +226,13 @@ function Operaciones() {
               </button>
             )}
           </div>
+          <JobProgressBar job={activeJob} />
           <pre className="ops-log">{activeJob.log || "Iniciando…"}</pre>
         </div>
       )}
 
+      <ReplicationCard />
+
       {/* Ingest folder */}
       <div className="card" style={{ padding: 20, marginBottom: 20 }}>
         <h2 className="section-title">Carpeta de ingesta</h2>
@@ -596,3 +602,219 @@ function OpTile({
     </div>
   );
 }
+
+/**
+ * Health of the read replica behind my.jorgecuadros.com.
+ *
+ * Worth a panel because the failure mode is silent: a replica whose SQL thread
+ * has stopped keeps answering queries, just with data frozen at the moment it
+ * stopped. Nothing on the customer site looks wrong — the balances are simply
+ * out of date — so without this the only signal is a customer complaining.
+ */
+function ReplicationCard() {
+  const [status, setStatus] = useState<ReplicationStatus | null>(null);
+  const [failed, setFailed] = useState(false);
+
+  const load = useCallback(() => {
+    getReplicationStatus()
+      .then((s) => {
+        setStatus(s);
+        setFailed(false);
+      })
+      .catch(() => setFailed(true));
+  }, []);
+
+  useEffect(() => {
+    load();
+    const t = setInterval(load, 30_000);
+    return () => clearInterval(t);
+  }, [load]);
+
+  // Not configured is the normal state in dev and before cutover, so it is a
+  // quiet note rather than an alarm — showing red here would train people to
+  // ignore the card.
+  if (failed || (status && !status.configured)) {
+    return (
+      <div className="card" style={{ padding: 20, marginBottom: 20 }}>
+        <h2 className="section-title">Réplica del sitio de clientes</h2>
+        <p className="inline-form-note">
+          {failed
+            ? "No se pudo consultar el estado de la réplica."
+            : "No configurada en este entorno."}
+        </p>
+      </div>
+    );
+  }
+
+  if (!status) {
+    return (
+      <div className="card" style={{ padding: 20, marginBottom: 20 }}>
+        <h2 className="section-title">Réplica del sitio de clientes</h2>
+        <p className="inline-form-note">Consultando…</p>
+      </div>
+    );
+  }
+
+  return (
+    <div className="card" style={{ padding: 20, marginBottom: 20 }}>
+      <div className="row-actions" style={{ justifyContent: "space-between" }}>
+        <h2 className="section-title" style={{ margin: 0 }}>
+          Réplica del sitio de clientes{" "}
+          <span className={`badge ${status.healthy ? "badge-positive" : "badge-negative"}`}>
+            {status.healthy ? "Replicando" : "Detenida"}
+          </span>
+        </h2>
+        <button className="btn btn-ghost" type="button" onClick={load}>
+          Actualizar
+        </button>
+      </div>
+
+      {status.problem && (
+        <div className="state-box state-error" style={{ marginTop: 12 }}>
+          {status.problem}
+        </div>
+      )}
+
+      <div className="kv-grid" style={{ paddingLeft: 0, paddingRight: 0 }}>
+        <KV label="Servidor" value={status.host} />
+        <KV label="Origen" value={status.sourceHost} />
+        <KV label="Hilo de E/S" value={status.ioRunning} />
+        <KV label="Hilo SQL" value={status.sqlRunning} />
+        {/* Never render a null lag as "0 s": MySQL reports NULL whenever a
+            thread is down, so the honest word is "unknown", not "up to date". */}
+        <KV
+          label="Retraso"
+          value={status.secondsBehind === null ? "sin dato" : `${status.secondsBehind} s`}
+        />
+        <KV label="Pendiente de aplicar" value={backlogLabel(status.apply)} />
+        <KV label="Consultado" value={formatDateTime(status.checkedAt)} />
+      </div>
+
+      <ApplyProgressBar apply={status.apply} />
+    </div>
+  );
+}
+
+/**
+ * Bytes the replica has fetched but not yet applied.
+ *
+ * Kept separate from the lag figure because it answers a question the lag
+ * cannot: while the SQL thread chews through one big transaction, the seconds
+ * counter can hold still, but this number visibly falls.
+ */
+function backlogLabel(apply: ApplyProgress | null): string {
+  if (!apply) return "sin dato";
+  // Different source binlog files means the replica is whole files behind and
+  // the byte delta is not a delta at all — positions restart in each new file.
+  if (!apply.sameFile) return "más de un archivo de binlog";
+  if (apply.backlogBytes === 0) return "al día";
+  return formatBytes(apply.backlogBytes);
+}
+
+/**
+ * Applied-vs-fetched bar. Rendered only when both threads are on the same
+ * source binlog file, because that is the only case where the percentage is
+ * arithmetic rather than a guess.
+ */
+function ApplyProgressBar({ apply }: { apply: ApplyProgress | null }) {
+  if (!apply || !apply.sameFile || apply.percent === null) return null;
+
+  return (
+    <div className="upload-progress" style={{ marginTop: 12 }}>
+      <div
+        className="progress-track"
+        role="progressbar"
+        aria-valuenow={apply.percent}
+        aria-valuemin={0}
+        aria-valuemax={100}
+        aria-label="Eventos aplicados de los recibidos"
+      >
+        <div className="progress-fill" style={{ width: `${apply.percent}%` }} />
+      </div>
+      <div className="upload-progress-stats mono">
+        <span>{apply.percent}% aplicado</span>
+        <span>
+          {apply.sourceLogFile} · {apply.execPos.toLocaleString("es-MX")} /{" "}
+          {apply.readPos.toLocaleString("es-MX")}
+        </span>
+      </div>
+    </div>
+  );
+}
+
+/** Matches the KV in the clientes/polizas/servicios detail pages. */
+function KV({ label, value }: { label: string; value: string | null | undefined }) {
+  return (
+    <div>
+      <div className="kv-label">{label}</div>
+      <div className="kv-value">{value || "—"}</div>
+    </div>
+  );
+}
+
+/**
+ * Step progress for a running migration.
+ *
+ * Only REIMPORT and SYNC report steps; BACKUP and RESTORE are a single
+ * mysqldump, so they render nothing here rather than a made-up bar — the
+ * spinner in the heading already says "working".
+ *
+ * The safety backup runs before the migration, so `progress` is null for the
+ * first stretch of every REIMPORT. That phase is named explicitly instead of
+ * showing 0%, which would read as "stuck".
+ */
+function JobProgressBar({ job }: { job: OpsJob }) {
+  const running = job.status === "RUNNING";
+  const p = job.progress;
+
+  if (!p) {
+    if (!running) return null;
+    return (
+      <p className="inline-form-note" style={{ marginTop: 8 }}>
+        Respaldo de seguridad previo…
+      </p>
+    );
+  }
+
+  return (
+    <div style={{ marginTop: 10, marginBottom: 4 }}>
+      <div
+        className="row-actions"
+        style={{ justifyContent: "space-between", marginBottom: 6 }}
+      >
+        <span className="inline-form-note" style={{ margin: 0 }}>
+          Paso {p.step} de {p.total} — {p.name}
+        </span>
+        <span className="inline-form-note" style={{ margin: 0 }}>
+          {p.percent}%
+        </span>
+      </div>
+      <div
+        role="progressbar"
+        aria-valuenow={p.percent}
+        aria-valuemin={0}
+        aria-valuemax={100}
+        aria-label={`Paso ${p.step} de ${p.total}`}
+        style={{
+          height: 6,
+          borderRadius: 999,
+          background: "var(--line)",
+          overflow: "hidden",
+        }}
+      >
+        <div
+          style={{
+            width: `${p.percent}%`,
+            height: "100%",
+            borderRadius: 999,
+            transition: "width 400ms ease",
+            background:
+              job.status === "FAILED"
+                ? "var(--negative)"
+                : "var(--positive)",
+          }}
+        />
+      </div>
+    </div>
+  );
+}

apps/web/src/lib/api.ts

@@ -59,6 +59,7 @@ import type {
   LookupsResponse,
   OpsJob,
   OpsJobKind,
+  ReplicationStatus,
   IngestFile,
   BackupFile,
   PropertyDetail,
@@ -939,6 +940,16 @@ export function deleteBackup(name: string): Promise<unknown> {
   return apiFetch(`/ops/backups/${encodeURIComponent(name)}`, { method: "DELETE" });
 }
 
+/**
+ * Health of the read replica my.jorgecuadros.com serves customers from.
+ *
+ * A stopped replica does not error — it answers with stale balances — so this
+ * is the only place the failure is visible.
+ */
+export function getReplicationStatus(): Promise<ReplicationStatus> {
+  return apiFetch<ReplicationStatus>("/ops/replication");
+}
+
 export function listOpsJobs(): Promise<OpsJob[]> {
   return apiFetch<OpsJob[]>("/ops/jobs");
 }

apps/web/src/lib/types.ts

@@ -61,6 +61,14 @@ export interface UserRow {
 export type OpsJobKind = "BACKUP" | "RESTORE" | "REIMPORT" | "SYNC";
 export type OpsJobStatus = "RUNNING" | "SUCCESS" | "FAILED";
 
+/** Derived from the job log by the API; null for jobs with no step markers. */
+export interface JobProgress {
+  step: number;
+  total: number;
+  name: string;
+  percent: number;
+}
+
 export interface OpsJob {
   id: string;
   kind: OpsJobKind;
@@ -70,6 +78,51 @@ export interface OpsJob {
   createdById: string | null;
   startedAt: string;
   finishedAt: string | null;
+  /** Only present on getOpsJob (the polled endpoint), not on the list. */
+  progress?: JobProgress | null;
+}
+
+/**
+ * Health of the MySQL read replica that my.jorgecuadros.com queries.
+ *
+ * `secondsBehind` is null whenever MySQL reports NULL, which it does when
+ * EITHER thread is down — so null means "unknown", never "up to date". Read
+ * `healthy`/`problem` rather than inferring health from the lag.
+ */
+export interface ReplicationStatus {
+  configured: boolean;
+  healthy: boolean;
+  host: string | null;
+  ioRunning: string | null;
+  sqlRunning: string | null;
+  secondsBehind: number | null;
+  lastIoError: string | null;
+  lastSqlError: string | null;
+  sourceHost: string | null;
+  apply: ApplyProgress | null;
+  problem: string | null;
+  checkedAt: string;
+}
+
+/**
+ * Relay-log apply progress, in source binlog bytes.
+ *
+ * Answers "is it moving?" when `secondsBehind` cannot: the lag counter sits
+ * still while the SQL thread works through one large transaction, but the
+ * backlog visibly shrinks. `backlogBytes === 0` is the only reading that means
+ * caught up — `percent` deliberately stops at 99.99 while bytes remain.
+ *
+ * Null fields when the two threads are on different source binlog files
+ * (`sameFile === false`), because the positions are then not comparable.
+ */
+export interface ApplyProgress {
+  sourceLogFile: string | null;
+  readPos: number;
+  relayLogFile: string | null;
+  execPos: number;
+  sameFile: boolean;
+  backlogBytes: number | null;
+  percent: number | null;
 }
 
 /** One of the four legacy Access files expected in the ingest folder. */

deploy/galactus/jorgecuadros-app.compose.yml

@@ -69,6 +69,14 @@ services:
       # apps/api/src/ops/ops.service.ts.
       OPS_DB_ADMIN_USER: ${OPS_DB_ADMIN_USER:-root}
       OPS_DB_ADMIN_PASSWORD: ${OPS_DB_ADMIN_PASSWORD:?OPS_DB_ADMIN_PASSWORD must be set}
+      # Read-only replica that my.jorgecuadros.com serves customers from. Used
+      # ONLY to report health on the Operaciones screen — the account holds
+      # REPLICATION CLIENT and nothing else, so it cannot read a single row.
+      # Unset is a supported state: the panel then says "no configurada"
+      # instead of erroring, which is correct before cutover and in dev.
+      REPLICA_DB_HOST: ${REPLICA_DB_HOST:-}
+      REPLICA_DB_USER: ${REPLICA_DB_USER:-}
+      REPLICA_DB_PASS: ${REPLICA_DB_PASS:-}
       S3_ENDPOINT: ${S3_ENDPOINT:?S3_ENDPOINT must be set}
       S3_BUCKET: ${S3_BUCKET:-jorgecuadros-documents}
       MINIO_ROOT_USER: ${MINIO_ROOT_USER:?MINIO_ROOT_USER must be set}

deploy/scripts/check-replication.sh

@@ -0,0 +1,85 @@
+#!/usr/bin/env bash
+#
+# Is the my.jorgecuadros.com read replica actually replicating?
+#
+#   deploy/scripts/check-replication.sh
+#
+# Answers it from the REPLICA alone, so it needs no credentials for the
+# galactus master — only ssh to the VPS. Exits non-zero when replication is
+# broken or lagging, so it is usable from cron or a monitor.
+#
+# Why not just eyeball `SHOW REPLICA STATUS`: the two obvious fields are both
+# misleading on their own.
+#
+#   * "Replica_IO_Running: Yes" only means the network thread is alive. The SQL
+#     thread can be stopped with a duplicate-key error while IO keeps happily
+#     downloading binlog, so the replica looks busy and falls further behind.
+#
+#   * "Seconds_Behind_Source: 0" reads 0 both when there is genuinely nothing
+#     to apply AND when the IO thread is disconnected — there is no event to
+#     measure staleness against, so absence of work is reported as being current.
+#
+# The trustworthy check is GTID_SUBTRACT(Retrieved, Executed): binlog we have
+# fetched but not yet applied. Empty means genuinely caught up.
+
+set -uo pipefail
+
+REPLICA_HOST="${REPLICA_HOST:-opc@163.192.62.37}"
+MAX_LAG="${MAX_LAG:-30}"
+
+raw=$(ssh -o ConnectTimeout=10 -o BatchMode=yes "$REPLICA_HOST" \
+  'sudo mysql -e "SHOW REPLICA STATUS\G"' 2>/dev/null)
+
+if [ -z "$raw" ]; then
+  echo "FAIL: could not reach $REPLICA_HOST or mysql returned nothing"
+  exit 2
+fi
+
+# sed rather than `head -n1`: on some machines `head` is shadowed by LWP's
+# HTTP head(1), which silently mangles the pipeline instead of erroring.
+field() { printf '%s\n' "$raw" | grep -E "^[[:space:]]*$1:" | sed -n '1p' | sed -E "s/^[[:space:]]*$1:[[:space:]]*//"; }
+
+io=$(field Replica_IO_Running)
+sql=$(field Replica_SQL_Running)
+lag=$(field Seconds_Behind_Source)
+io_err=$(field Last_IO_Error)
+sql_err=$(field Last_SQL_Error)
+
+# The authoritative "am I caught up" test: anything fetched but not applied.
+backlog=$(ssh -o ConnectTimeout=10 -o BatchMode=yes "$REPLICA_HOST" \
+  'sudo mysql -NB -e "
+     SELECT IFNULL(NULLIF(GTID_SUBTRACT(
+       (SELECT RECEIVED_TRANSACTION_SET FROM performance_schema.replication_connection_status),
+       @@GLOBAL.gtid_executed), \"\"), \"(none)\")" 2>/dev/null' 2>/dev/null)
+[ -z "$backlog" ] && backlog="(performance_schema off — using lag only)"
+
+echo "replica     : $REPLICA_HOST"
+echo "IO thread   : $io"
+echo "SQL thread  : $sql"
+if [ "$lag" = "NULL" ] || [ -z "$lag" ]; then
+  echo "lag         : NULL"
+else
+  echo "lag         : ${lag}s"
+fi
+echo "unapplied   : $backlog"
+[ -n "$io_err" ]  && echo "IO error    : $io_err"
+[ -n "$sql_err" ] && echo "SQL error   : $sql_err"
+
+rc=0
+[ "$io" = "Yes" ]  || { echo "FAIL: IO thread not running"; rc=1; }
+[ "$sql" = "Yes" ] || { echo "FAIL: SQL thread not running"; rc=1; }
+[ -n "$io_err" ]   && { rc=1; }
+[ -n "$sql_err" ]  && { rc=1; }
+# SHOW reports NULL lag whenever EITHER thread is down — there is no applied
+# event to measure against. Never report which one from the lag alone; the
+# thread fields above already said, and guessing produces a wrong diagnosis.
+if [ "$lag" = "NULL" ] || [ -z "$lag" ]; then
+  echo "FAIL: lag is NULL (replication not applying)"
+  rc=1
+elif [ "$lag" -gt "$MAX_LAG" ] 2>/dev/null; then
+  echo "WARN: lag ${lag}s exceeds ${MAX_LAG}s"
+  rc=1
+fi
+
+[ $rc -eq 0 ] && echo "OK: replica is running and caught up"
+exit $rc

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).
 
@@ -78,7 +83,13 @@ SYNC_STEPS = [
 ]
 
 
-def run(cmd: list[str]) -> None:
+def run(cmd: list[str], step: int | None = None, total: int | None = None) -> None:
+    # The "[paso i/N] name" marker is a contract with the Operaciones screen,
+    # which parses the last one to show progress. Emitting it here rather than
+    # letting the UI count STEPS itself keeps the two from drifting when a step
+    # is added — the number of steps is only ever stated in this file.
+    if step is not None and total is not None:
+        print(f"[paso {step}/{total}] {Path(cmd[1]).name}", flush=True)
     print("+ " + " ".join(cmd), flush=True)
     r = subprocess.run(cmd)
     if r.returncode:
@@ -94,14 +105,22 @@ def main() -> None:
                     help="upsert legacy rows and archive removed legacy rows; preserve manual rows")
     args = ap.parse_args()
 
-    if args.stage:
+    steps = SYNC_STEPS if args.sync else STEPS
-        run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")])
+    # 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
 
-    for step in SYNC_STEPS if args.sync else STEPS:
+    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)
+        run(cmd, step=i, total=total)
 
     print(f"\n✓ migration complete for env={args.env}")
 

migration/transform_customers.py

@@ -189,6 +189,11 @@ def customer_from_utilities(row, name_index) -> dict:
         customerSince=as_date(row["cliente_desde"]),
         status=as_bool(row["status"]),
         feeAmount=as_decimal(row["fee"]),
+        # DATGRAL.TIPO is the minimum-balance threshold (100/200/300/500 —
+        # 1,017 of 1,172 customers carry one), NOT an identification or account
+        # type as the column name suggests. It reaches the website as
+        # datosfreak.TIPO and is returned to the customer app as `minBalance`.
+        minimumBalance=as_decimal(row["tipo"]),
         updatedAt=NOW,
     )
 
@@ -217,6 +222,7 @@ def customer_from_insurance(row, name_index) -> dict:
         customerSince=None,
         status=1,
         feeAmount=None,
+        minimumBalance=None,
         updatedAt=NOW,
     )
 
@@ -225,7 +231,7 @@ _CUST_COLS = [
     "id", "name", "nameSource", "nameMissing", "addressLine1", "addressLine2", "city", "state", "zipCode",
     "country", "phone", "mobile", "fax", "email", "notes", "identificationType",
     "identificationNumber", "identificationExpiration", "customerSince",
-    "status", "feeAmount", "updatedAt",
+    "status", "feeAmount", "minimumBalance", "updatedAt",
 ]
 
 
@@ -316,7 +322,7 @@ def main() -> None:
             remap[rec["id"]] = stable or rec["id"]
         for rec in customers:
             rec["id"] = remap[rec["id"]]
-            cur.execute(f"INSERT INTO customers ({','.join(f'`{c}`' for c in _CUST_COLS)}) VALUES ({placeholders}) ON DUPLICATE KEY UPDATE name=VALUES(name),nameSource=VALUES(nameSource),nameMissing=VALUES(nameMissing),addressLine1=VALUES(addressLine1),addressLine2=VALUES(addressLine2),city=VALUES(city),state=VALUES(state),zipCode=VALUES(zipCode),country=VALUES(country),phone=VALUES(phone),mobile=VALUES(mobile),fax=VALUES(fax),email=VALUES(email),notes=VALUES(notes),identificationType=VALUES(identificationType),identificationNumber=VALUES(identificationNumber),identificationExpiration=VALUES(identificationExpiration),customerSince=VALUES(customerSince),status=VALUES(status),feeAmount=VALUES(feeAmount),updatedAt=VALUES(updatedAt)", tuple(rec[c] for c in _CUST_COLS))
+            cur.execute(f"INSERT INTO customers ({','.join(f'`{c}`' for c in _CUST_COLS)}) VALUES ({placeholders}) ON DUPLICATE KEY UPDATE name=VALUES(name),nameSource=VALUES(nameSource),nameMissing=VALUES(nameMissing),addressLine1=VALUES(addressLine1),addressLine2=VALUES(addressLine2),city=VALUES(city),state=VALUES(state),zipCode=VALUES(zipCode),country=VALUES(country),phone=VALUES(phone),mobile=VALUES(mobile),fax=VALUES(fax),email=VALUES(email),notes=VALUES(notes),identificationType=VALUES(identificationType),identificationNumber=VALUES(identificationNumber),identificationExpiration=VALUES(identificationExpiration),customerSince=VALUES(customerSince),status=VALUES(status),feeAmount=VALUES(feeAmount),minimumBalance=VALUES(minimumBalance),updatedAt=VALUES(updatedAt)", tuple(rec[c] for c in _CUST_COLS))
         for ref in refs:
             cur.execute("INSERT INTO customer_legacy_refs (id,customerId,sourceSystem,sourceTable,legacyId) VALUES (%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId)",
                         (ref[0], remap[ref[1]], ref[2], ref[3], ref[4]))

migration/transform_transactions.py

@@ -112,6 +112,29 @@ def main():
         type_rows.append((tid, en, s(r["espa_ol"]), 0))
         type_map[en.upper()] = tid
 
+    def type_id_for(raw) -> str | None:
+        """Resolve a transaction type, minting one when the lookup lacks it.
+
+        The Access `TYPE OF TRX` table is a stale pick-list, not a constraint —
+        staff free-text straight into DATOS2, so 78 values covering 3,939 rows
+        (BALANCE FORWARD 1,188, ANNUAL FEE 1,116, IZZI 367, ...) appear in the
+        ledger but not the lookup. Leaving those unmapped stored typeId NULL and
+        lost the label outright: nothing else on `transactions` carries the type
+        text, so the row rendered blank and was unrecoverable after migration.
+        Minting from the literal keeps the display string; nameEs stays NULL
+        because only the lookup has translations.
+        """
+        en = s(raw)
+        if not en:
+            return None
+        key = en.upper()
+        tid = type_map.get(key)
+        if tid is None:
+            tid = str(uuid.uuid4())
+            type_rows.append((tid, en, None, 0))
+            type_map[key] = tid
+        return tid
+
     xr = load("stg_utilities", "tipo_hist")
     xr_rows = []
     for _, r in xr.iterrows():
@@ -126,10 +149,11 @@ def main():
     skip_cust = skip_date = skip_dupe = 0
 
     def add(cid, domain, tdate, amount, currency, *, period=None, reference=None,
-            typeid=None, check=None, message=None, src_db=None, src_tbl=None, legacy=None):
+            typeid=None, check=None, message=None, src_db=None, src_tbl=None, legacy=None,
+            outstanding=0):
         tx.append((str(uuid.uuid4()), cid, domain, typeid, tdate, period, reference,
                    amount if amount is not None else Decimal(0), currency, None, check,
-                   message, 0, src_db, src_tbl, legacy))
+                   message, outstanding, src_db, src_tbl, legacy))
 
     # Business key of a real cash payment. `folio` is deliberately excluded: it
     # is a per-table sequential number that collides between EFECTIVO and
@@ -143,12 +167,24 @@ def main():
             s(r["conepto"]),
         )
 
-    def efectivo_like(src, name, domain, custmap, src_db, legacy_tbl, *, seen=None):
+    def efectivo_like(src, name, domain, custmap, src_db, legacy_tbl, *, seen=None,
+                      type_label=None):
         """Load an EFECTIVO-shaped cash ledger.
 
         `seen` (a set) makes the load de-duplicating: keys are added to it as
         rows load, and a row whose key is already present is skipped. That is
         how EFECTIVO_BACKUP contributes only its genuinely-new rows.
+
+        `type_label` names the transaction type for every row. These tables have
+        no type column at all — in Access the type is implied by which table the
+        row lives in — so unlike DATOS2 there is no string to map and typeId came
+        out NULL for all of them.
+
+        That is not merely a blank label. handleGetAccountDetails in
+        my.jorgecuadros.com identifies payments by matching TYPEOFTRX against
+        ('PAYMENT THANK YOU', 'PAYPAL', 'CASH DEPOSIT', 'CHECK DEPOSIT') to reset
+        the running balance in mode=current; an unlabelled payment is not
+        recognised and the balance silently diverges from legacy.
         """
         nonlocal skip_cust, skip_date, skip_dupe
         df = load(src, name)
@@ -166,9 +202,20 @@ def main():
                 skip_date += 1; continue
             add(cid, domain, td, dec(r["monto"], Decimal(0)), cur(r["monedas"]),
                 reference=s(r["folio"]), message=s(r["conepto"]),
+                typeid=type_id_for(type_label),
                 src_db=src_db, src_tbl=legacy_tbl, legacy=str(int(r["_row_num"])))
 
     def fm3(name, legacy_tbl, check_col=None):
+        """FM3 fee streams. Deliberately left unlabelled, unlike EFECTIVO.
+
+        These rows (EFECTIVO FM3 627, CHEQUE FM3 157) also have no type column,
+        but every one of them predates the two periods the site exposes — it
+        allowlists only the current year and the prior year — so none can be
+        matched against a legacy label, and none can reach a customer. Inventing
+        a plausible name like "CHECK DEPOSIT" would feed the payment-detection
+        list in handleGetAccountDetails on nothing but a guess. Leave them NULL
+        until a real mapping is available.
+        """
         nonlocal skip_cust, skip_date
         df = load("stg_utilities", name)
         for _, r in df.iterrows():
@@ -184,6 +231,16 @@ def main():
                 src_db="UTILITIES", src_tbl=legacy_tbl, legacy=str(int(r["_row_num"])))
 
     def billing(name, legacy_tbl):
+        """Load a DATOS2-shaped billing ledger.
+
+        NOPAGO is the legacy "still owed" flag. The website reads it directly —
+        `account.statement.php` splits the statement on `NOPAGO = 0` vs
+        `NOPAGO = 1` and renders the latter as the "Outstanding Bills Requiring
+        Attention" table — so dropping it does not merely lose a column, it
+        silently empties that whole section for anyone served off the platform.
+        Only these three tables carry it (76 rows set in DATOS2 today); the
+        EFECTIVO/FM3 cash streams have no such column and stay 0.
+        """
         nonlocal skip_cust, skip_date
         df = load("stg_utilities", name)
         for _, r in df.iterrows():
@@ -193,11 +250,12 @@ def main():
             td = dt(r["date"])
             if td is None:
                 skip_date += 1; continue
-            tid = type_map.get((s(r["type_of_trx"]) or "").upper())
+            tid = type_id_for(r["type_of_trx"])
             add(cid, "UTILITY", td, dec(r["chargecredit"], Decimal(0)), "MXN",
                 period=s(r["period"]), reference=s(r["refer"]), typeid=tid,
                 check=s(r["cheque"]), src_db="UTILITIES", src_tbl=legacy_tbl,
-                legacy=str(int(r["_row_num"])))
+                legacy=str(int(r["_row_num"])),
+                outstanding=1 if s(r["nopago"]) == "1" else 0)
 
     def iva():
         nonlocal skip_cust
@@ -214,17 +272,25 @@ def main():
     # order matters: EFECTIVO is the live table and loads first, so a collision
     # always resolves in its favour.
     cash_seen: set = set()
+    # "CASH DEPOSIT" is not a guess: matching these rows to the live site on
+    # (NUMid, date, amount) resolves to that label unanimously — 66/66 in the
+    # current-year `datosfreak` and 100/100 in the prior-year `2025` table,
+    # which are the only two periods the site exposes.
     efectivo_like("stg_utilities", "efectivo", "UTILITY", util_cust, "UTILITIES",
-                  "EFECTIVO", seen=cash_seen)
+                  "EFECTIVO", seen=cash_seen, type_label="CASH DEPOSIT")
     efectivo_like("stg_utilities", "efectivo_backup", "UTILITY", util_cust, "UTILITIES",
-                  "EFECTIVO_BACKUP", seen=cash_seen)
+                  "EFECTIVO_BACKUP", seen=cash_seen, type_label="CASH DEPOSIT")
     fm3("efectivo_fm3", "EFECTIVO FM3")
     fm3("cheque_fm3", "CHEQUE FM3", check_col="num_cheque")
     billing("datos2", "datos2")
     billing("fee_anual", "FEE ANUAL")
     billing("fee15", "fee15")
     iva()
-    efectivo_like("stg_seguros", "efectivo", "INSURANCE", ins_cust, "SEGUROS 16_be", "EFECTIVO")
+    # Same record shape in the seguros DB. Labelled for consistency in the
+    # platform's own UI; unverifiable against the site, which only ever reads
+    # domain='UTILITY', so no customer-facing behaviour depends on it.
+    efectivo_like("stg_seguros", "efectivo", "INSURANCE", ins_cust, "SEGUROS 16_be",
+                  "EFECTIVO", type_label="CASH DEPOSIT")
 
     if sync_mode:
         # Transaction types are rebuilt with fresh uuids each run; resolve them
@@ -243,7 +309,7 @@ def main():
         if new_types:
             c.executemany("INSERT INTO type_transactions (id,nameEn,nameEs,isService) VALUES (%s,%s,%s,%s)", new_types)
         tx = [(t[0], t[1], t[2], (db_types.get(fresh_name.get(t[3])) if t[3] else None), *t[4:]) for t in tx]
-        c.executemany("INSERT INTO transactions (id,customerId,domain,typeId,transactionDate,period,reference,amount,currency,exchangeRate,checkNumber,message,outstanding,legacySourceDb,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),domain=VALUES(domain),typeId=VALUES(typeId),transactionDate=VALUES(transactionDate),period=VALUES(period),reference=VALUES(reference),amount=VALUES(amount),currency=VALUES(currency),checkNumber=VALUES(checkNumber),message=VALUES(message),voidedAt=NULL", tx)
+        c.executemany("INSERT INTO transactions (id,customerId,domain,typeId,transactionDate,period,reference,amount,currency,exchangeRate,checkNumber,message,outstanding,legacySourceDb,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),domain=VALUES(domain),typeId=VALUES(typeId),transactionDate=VALUES(transactionDate),period=VALUES(period),reference=VALUES(reference),amount=VALUES(amount),currency=VALUES(currency),checkNumber=VALUES(checkNumber),message=VALUES(message),outstanding=VALUES(outstanding),voidedAt=NULL", tx)
     else:
         c.execute("SET FOREIGN_KEY_CHECKS=0")
         for t in ("transactions", "type_transactions", "exchange_rates"):

package.json

@@ -1,6 +1,6 @@
 {
   "name": "jorgecuadros-platform",
-  "version": "1.0.7",
+  "version": "1.0.15",
   "private": true,
   "workspaces": [
     "apps/*",

packages/database/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@jorgecuadros/database",
-  "version": "1.0.7",
+  "version": "1.0.15",
   "private": true,
   "main": "generated/client/index.js",
   "types": "generated/client/index.d.ts",