jorgecuadros-platform/apps/api/src/ops/replication.service.ts

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;
}

/**
 * How far the replica's executed history is from the master's, in transactions.
 *
 * This is the check `SHOW REPLICA STATUS` cannot give you, and it is stronger
 * than everything else on the card for one specific reason: every other field is
 * self-reported by the replica. `Seconds_Behind_Source` reads 0 both when there
 * is genuinely nothing to apply AND when the I/O thread is disconnected — with
 * no incoming event there is nothing to measure staleness against, so a dead
 * link reports as perfectly current. `GTID_SUBTRACT(master, replica)` asks the
 * master what it has done and the replica what it has applied, so a silent
 * disconnect shows up immediately as a growing number.
 */
export interface GtidDrift {
  /** Transactions the master executed that the replica has not. 0 = identical. */
  missingTransactions: number;
  /** The missing GTID set verbatim. Null when nothing is missing. */
  missingGtidSet: string | null;
  /**
   * Transactions in the replica's `gtid_executed` under its OWN server UUID —
   * writes that happened here and exist nowhere on the master.
   *
   * Reported, never alarmed on. A non-zero count is the expected residue of the
   * seed load: restoring a dump executes its statements locally, and they take
   * GTIDs from this server's UUID. They never propagate (`log_replica_updates`
   * is off and nothing sources from this node), so they are harmless — right up
   * until someone tries to promote this box, where they become a real divergence.
   */
  localTransactions: number;
}

/** One table's row count and content fingerprint, on one side of the link. */
export interface TableFingerprint {
  table: string;
  masterRows: number;
  replicaRows: number;
  /** Order-independent checksum over every column of every row. */
  masterChecksum: string;
  replicaChecksum: string;
  matches: boolean;
}

export interface VerifyResult {
  /** True only when every table matched on both count and checksum. */
  identical: boolean;
  tables: TableFingerprint[];
  /** Set instead of `tables` when the comparison could not be run at all. */
  problem: string | null;
  checkedAt: string;
  /** Wall-clock cost, because this is a full scan and the caller should see it. */
  elapsedMs: number;
}

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;
  /** GTID comparison against the master. Null when the master was unreachable. */
  drift: GtidDrift | null;
  /** Human-readable reason when healthy is false. */
  problem: string | null;
  checkedAt: string;
}

/**
 * The tables `my.jorgecuadros.com` reads through the `web_reader` grant.
 *
 * This list is the verification surface, not the replication surface — the
 * replica carries the whole schema. These are the eight whose divergence would
 * actually be visible to a customer, so they are the ones worth a full scan.
 */
export const REPLICATED_TABLES = [
  "transactions",
  "customers",
  "customer_legacy_refs",
  "type_transactions",
  "exchange_rates",
  "properties",
  "property_services",
  "trust_accounts",
] as const;

/**
 * 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,
      drift: null,
      problem: null,
      checkedAt: now,
    };

    if (!host || !user || !password) {
      return { ...empty, problem: "REPLICA_DB_* no configuradas" };
    }

    let raw: string;
    try {
      raw = await this.onReplica("SHOW REPLICA STATUS\\G");
    } 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),
      // Also reported rather than alarmed on, for the same reason: a busy master
      // is always a few transactions ahead for the instant they are in flight.
      // Null rather than zero when the master could not be reached — "unknown"
      // and "identical" must not render the same.
      drift: await this.gtidDrift(),
      problem,
      checkedAt: now,
    };
  }

  /**
   * Compare executed history between master and replica.
   *
   * Two round trips: ask the master what it has executed, then ask the replica
   * to subtract its own history from that. The subtraction runs on the replica
   * rather than in TypeScript because `GTID_SUBTRACT` already implements the
   * interval algebra correctly, and reimplementing set subtraction over binlog
   * ranges is exactly the kind of thing that looks right and is wrong at the
   * boundaries.
   *
   * @returns null on any failure — a broken drift check must never be mistaken
   *          for a healthy zero.
   */
  private async gtidDrift(): Promise<GtidDrift | null> {
    try {
      const masterGtid = (await this.onMaster("SELECT @@gtid_executed")).trim();

      // GTID sets are UUIDs, digits, colons, commas, hyphens, whitespace and
      // (since 8.4) alphanumeric tags. Nothing else is legal, so rejecting
      // anything outside that alphabet is a whitelist, not a blacklist: with no
      // quote and no backslash able to survive it, the value cannot escape the
      // string literal it is interpolated into below.
      if (masterGtid && !/^[0-9a-fA-F:,\s_-]+$/.test(masterGtid)) {
        this.logger.warn("gtid_executed del maestro con formato inesperado");
        return null;
      }
      // An empty set means the master has GTID mode off, and there is nothing
      // meaningful to compare.
      if (!masterGtid) return null;

      const flat = masterGtid.replace(/\s+/g, "");
      // Every GTID set is flattened with REPLACE before it leaves the server.
      // MySQL wraps `gtid_executed` across lines once it holds more than one
      // source UUID, and this is read back as tab-separated columns — an
      // embedded newline would split one row into two and silently truncate the
      // set at the first UUID.
      const out = await this.onReplica(
        "SELECT REPLACE(GTID_SUBTRACT(" +
          `'${flat}', @@gtid_executed), '\\n', ''), ` +
          "@@server_uuid, REPLACE(@@gtid_executed, '\\n', '')",
        ["-N"],
      );

      // Trailing newline only — never `.trim()`. When nothing is missing the
      // first column is the empty string, so the line begins with a tab, and
      // trimming it would shift every column one position left and report the
      // replica's own UUID as the missing GTID set.
      const [missingSet = "", serverUuid = "", executed = ""] = out
        .replace(/\r?\n+$/, "")
        .split("\t");

      return {
        missingTransactions: countGtids(missingSet),
        missingGtidSet: missingSet || null,
        localTransactions: countGtids(gtidsForUuid(executed, serverUuid)),
      };
    } catch (e) {
      const msg = e instanceof Error ? e.message : String(e);
      this.logger.warn(`no se pudo comparar GTIDs con el maestro: ${msg}`);
      return null;
    }
  }

  /**
   * Full-scan comparison of the customer-visible tables on both sides.
   *
   * Deliberately NOT part of `status()`: this reads every row of every table in
   * `REPLICATED_TABLES` on both servers, so it belongs behind a button, not a
   * 30-second poll.
   *
   * It answers the one question GTID drift cannot. GTIDs prove the replica
   * applied every transaction the master produced; they say nothing about rows
   * changed on the replica by some other route. A local write is invisible to
   * every other field on the card and shows up here as a checksum mismatch.
   */
  async verify(): Promise<VerifyResult> {
    const started = Date.now();
    const base: VerifyResult = {
      identical: false,
      tables: [],
      problem: null,
      checkedAt: new Date().toISOString(),
      elapsedMs: 0,
    };

    let sql: string;
    try {
      sql = await this.fingerprintSql();
    } catch (e) {
      const msg = e instanceof Error ? e.message : String(e);
      return { ...base, problem: `No se pudo leer el esquema: ${msg}`, elapsedMs: Date.now() - started };
    }

    let masterOut: string;
    let replicaOut: string;
    try {
      // Sequential, not parallel. Running both at once would have the master
      // scan under the replica's own read load only sometimes, which makes a
      // slow run hard to attribute; and the boxes are small enough that two
      // concurrent full scans is a real memory event on the 946MB replica.
      masterOut = await this.onMaster(sql);
      replicaOut = await this.onReplica(sql);
    } catch (e) {
      const msg = e instanceof Error ? e.message : String(e);
      return { ...base, problem: `No se pudo comparar: ${msg}`, elapsedMs: Date.now() - started };
    }

    const master = parseFingerprints(masterOut);
    const replica = parseFingerprints(replicaOut);

    const tables: TableFingerprint[] = REPLICATED_TABLES.map((table) => {
      const m = master.get(table);
      const r = replica.get(table);
      return {
        table,
        masterRows: m?.rows ?? -1,
        replicaRows: r?.rows ?? -1,
        masterChecksum: m?.checksum ?? "?",
        replicaChecksum: r?.checksum ?? "?",
        // Both sides must have answered. A missing row on either side is a
        // mismatch, never a pass — `undefined === undefined` would otherwise
        // report two failed reads as agreement.
        matches:
          m !== undefined && r !== undefined && m.rows === r.rows && m.checksum === r.checksum,
      };
    });

    return {
      identical: tables.every((t) => t.matches),
      tables,
      problem: null,
      checkedAt: base.checkedAt,
      elapsedMs: Date.now() - started,
    };
  }

  /**
   * Build the count+checksum query from the live column list.
   *
   * The columns come from `information_schema` on the master rather than being
   * hardcoded, so the check keeps covering the whole row after a migration adds
   * one. Reading the schema from the master is safe by construction: if the two
   * schemas had diverged, replication would already be broken.
   */
  private async fingerprintSql(): Promise<string> {
    const list = REPLICATED_TABLES.map((t) => `'${t}'`).join(",");
    const raw = await this.onMaster(
      "SELECT CONCAT(TABLE_NAME, '\\t', COLUMN_NAME) FROM information_schema.COLUMNS " +
        `WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME IN (${list}) ` +
        "ORDER BY TABLE_NAME, ORDINAL_POSITION",
      ["-N"],
    );

    const cols = new Map<string, string[]>();
    for (const line of raw.split("\n")) {
      const [table, column] = line.trim().split("\t");
      if (!table || !column) continue;
      cols.set(table, [...(cols.get(table) ?? []), column]);
    }

    const selects = REPLICATED_TABLES.map((table) => {
      const columns = cols.get(table);
      if (!columns?.length) throw new Error(`tabla ${table} sin columnas`);
      // CONVERT(... USING binary), never CAST(... AS CHAR).
      //
      // CAST to CHAR transcodes into the *connection* character set, which is
      // not the same on the two servers: the mysql client inside the master's
      // container negotiates latin1, while the replica's negotiates utf8mb4.
      // Every accented character in a Mexican name, street or note therefore
      // hashes to different bytes on each side, and the comparison reports a
      // permanent mismatch on exactly the tables that hold free text — a
      // verification tool that always cries wolf, which is worse than none.
      // Comparing the stored bytes sidesteps the session entirely. (Verified
      // 2026-08-06: with CAST, `customers.name` gave 3344437324815 vs
      // 3339150372121; with CONVERT both give 3339150372121.)
      //
      // 0x1f (unit separator) joins the columns and 0x1e (record separator)
      // stands in for NULL. Both matter: CONCAT_WS *skips* NULLs rather than
      // emitting an empty field, so without a placeholder the rows
      // ('a', NULL, 'b') and ('a', 'b', NULL) produce the same string and a
      // column-shifting bug would checksum as identical.
      const expr = columns
        .map((c) => `IFNULL(CONVERT(\`${c}\` USING binary), 0x1e)`)
        .join(", 0x1f, ");
      // SUM, not a running hash: addition is commutative, so the result does not
      // depend on the order rows come back in. The two servers have no reason to
      // scan in the same order and are not asked to.
      return (
        `SELECT '${table}' AS t, COUNT(*) AS n, ` +
        `IFNULL(SUM(CRC32(CONCAT_WS(0x1f, ${expr}))), 0) AS c FROM \`${table}\``
      );
    });

    return selects.join(" UNION ALL ");
  }

  /* ------------------------------------------------------------ plumbing */

  /**
   * Run a statement on the replica.
   *
   * --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 the tailnet ACL 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.
   */
  private async onReplica(sql: string, extra: string[] = []): Promise<string> {
    const host = process.env.REPLICA_DB_HOST!;
    const user = process.env.REPLICA_DB_USER!;
    const password = process.env.REPLICA_DB_PASS!;
    const { stdout } = await exec(
      "mysql",
      [
        `--host=${host}`,
        `--user=${user}`,
        "--ssl",
        "--ssl-verify-server-cert=0",
        "--connect-timeout=5",
        ...extra,
        "-e",
        sql,
      ],
      { env: { ...process.env, MYSQL_PWD: password }, timeout: VERIFY_TIMEOUT_MS },
    );
    return stdout;
  }

  /**
   * Run a statement on the master, using the application's own DATABASE_URL.
   *
   * The app credential is enough here on purpose — everything this class sends
   * to the master is a SELECT against `information_schema` or a system variable.
   * Reaching for OPS_DB_ADMIN_* the way OpsService does would hand a monitoring
   * read path a credential that can also restore a dump.
   */
  private async onMaster(sql: string, extra: string[] = []): Promise<string> {
    const raw = process.env.DATABASE_URL;
    if (!raw) throw new Error("DATABASE_URL no está configurada");
    const u = new URL(raw);
    const { stdout } = await exec(
      "mysql",
      [
        `--host=${u.hostname}`,
        `--port=${u.port || "3306"}`,
        `--user=${decodeURIComponent(u.username)}`,
        "--connect-timeout=5",
        ...extra,
        "-N",
        "-e",
        sql,
        u.pathname.replace(/^\//, ""),
      ],
      {
        env: { ...process.env, MYSQL_PWD: decodeURIComponent(u.password) },
        timeout: VERIFY_TIMEOUT_MS,
      },
    );
    return stdout;
  }
}

/** Full scans on a 1-vCPU replica are not fast; 15s would cut them off. */
const VERIFY_TIMEOUT_MS = 120_000;

/** Parse the `t\tn\tc` rows the fingerprint query emits under `mysql -N`. */
function parseFingerprints(raw: string): Map<string, { rows: number; checksum: string }> {
  const out = new Map<string, { rows: number; checksum: string }>();
  for (const line of raw.split("\n")) {
    const [table, n, c] = line.trim().split("\t");
    if (!table || n === undefined || c === undefined) continue;
    const rows = Number(n);
    if (!Number.isFinite(rows)) continue;
    // The checksum stays a string. Sums of CRC32 over 40k rows exceed 2^53, so
    // parsing it as a number would round and make distinct tables compare equal.
    out.set(table, { rows, checksum: c });
  }
  return out;
}

/**
 * Count the transactions in a GTID set.
 *
 * Exported for testing. The format is `uuid[:tag]:interval[:interval]...`,
 * comma-separated, where an interval is `N` or `N-M` inclusive at both ends —
 * so `1-5` is five transactions, not four.
 *
 * MySQL 8.4 added an optional alphanumeric tag between the UUID and the first
 * interval. It is skipped rather than parsed: any segment that is not a number
 * or a number range is not an interval, whatever else it may be.
 */
export function countGtids(set: string): number {
  if (!set.trim()) return 0;
  let total = 0;
  for (const group of set.split(",")) {
    for (const part of group.trim().split(":").slice(1)) {
      const m = /^(\d+)(?:-(\d+))?$/.exec(part.trim());
      if (!m) continue;
      const from = Number(m[1]);
      const to = m[2] === undefined ? from : Number(m[2]);
      if (Number.isFinite(from) && Number.isFinite(to) && to >= from) total += to - from + 1;
    }
  }
  return total;
}

/**
 * Narrow a GTID set to the intervals belonging to one server UUID.
 *
 * Exported for testing. Used to isolate the replica's own writes from the
 * history it replicated, which are interleaved in the same `gtid_executed`.
 */
export function gtidsForUuid(set: string, uuid: string): string {
  if (!uuid.trim()) return "";
  const wanted = uuid.trim().toLowerCase();
  return set
    .split(",")
    .map((g) => g.trim())
    .filter((g) => g.toLowerCase().startsWith(`${wanted}:`))
    .join(",");
}

/**
 * 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;
}